from dataclasses import dataclass from campaigns.auth.dtos import User from campaigns.campaigns.enums import CampaignStatus from campaigns.connectors.db import Database, session_context from campaigns.connectors.facebook.arguments import ( AdCreativeArgs, AdCreativeInstagramPostArgs, CreateAdCreativeArgs, CreateAdsetArgs, UpdateAdArgs, ) from campaigns.connectors.facebook.base import FacebookClient from campaigns.connectors.facebook.enums import AdSetBillingEvent, ObjectStatus from campaigns.connectors.facebook.models import ( AdCampaign, AdSetCustomAudienceTargeting, AdSetPosition, AdSetTargetingAudienceTargeting, CustomAudience, ) from campaigns.core.handler import Handler from campaigns.meta.enums import CampaignContentType, CampaignPlatform from campaigns.meta.models import Campaign, EngagementAudience, TargetingAudience from campaigns.meta.models.campaign_ad import CampaignAd from campaigns.meta.models.campaign_ad_creative import CampaignAdCreative from campaigns.meta.models.campaign_ad_set import CampaignAdSet from campaigns.meta.models.campaign_ad_set_engagement_audience import ( CampaignAdSetEngagementAudience, ) from campaigns.meta.models.campaign_ad_set_targeting_audience import ( CampaignAdSetTargetingAudience, ) from campaigns.meta.repositories import ( CampaignAdCreativeRepository, CampaignAdRepository, CampaignAdSetEngagementAudienceRepository, CampaignAdSetRepository, CampaignAdSetTargetingAudienceRepository, CampaignRepository, ) from campaigns.meta.services import AdFormatService, CampaignService from campaigns.meta.validators import CampaignValidator @dataclass(frozen=True) class SubmitCampaignRequest: user: User campaign_id: str @dataclass(frozen=True) class SubmitCampaignResponse: status: CampaignStatus @dataclass class SubmitCampaignHandler( Handler[SubmitCampaignRequest, SubmitCampaignResponse], ): db: Database facebook_client: FacebookClient campaign_service: CampaignService campaign_validator: CampaignValidator campaign_repository: CampaignRepository campaign_ad_set_repository: CampaignAdSetRepository campaign_ad_set_targeting_audience_repository: CampaignAdSetTargetingAudienceRepository campaign_ad_set_engagement_audience_repository: CampaignAdSetEngagementAudienceRepository campaign_ad_creative_repository: CampaignAdCreativeRepository campaign_ad_repository: CampaignAdRepository ad_format_service: AdFormatService @session_context(in_transaction=True) async def handle(self, request: SubmitCampaignRequest) -> SubmitCampaignResponse: campaign = await self.campaign_service.get_user_campaign( user=request.user, campaign_id=request.campaign_id, ) # !TODO add specific validations for each CampaignPlatform await self.campaign_validator.validate( campaign, user=request.user, with_external=True, ) ad_campaign = await self.create_ad_campaign(campaign) ad_creative = await self.create_ad_creative(campaign) ads = [] excluded_custom_audiences = [] if campaign.engagement_audience: engagement_ad_set = await self.create_engagement_audience_ad_set( campaign, engagement_audience=campaign.engagement_audience, campaign_external_id=ad_campaign.id, ) ad = await self.create_ad( campaign, ad_set=engagement_ad_set, ad_creative=ad_creative, ) ads.append(ad) excluded_custom_audiences.append( CustomAudience(id=campaign.engagement_audience.external_id) ) if campaign.targeting_audience: targeting_ad_set = await self.create_targeting_audience_ad_set( campaign, targeting_audience=campaign.targeting_audience, campaign_external_id=ad_campaign.id, excluded_custom_audiences=excluded_custom_audiences, ) ad = await self.create_ad( campaign, ad_set=targeting_ad_set, ad_creative=ad_creative, ) ads.append(ad) await self.submit_ad_campaign(ads=ads) campaign.external_id = ad_campaign.id campaign.status = CampaignStatus.SUBMITTED self.campaign_repository.add(campaign) return SubmitCampaignResponse(status=campaign.status) async def create_ad_campaign(self, campaign: Campaign) -> AdCampaign: if not campaign.lifetime_budget: raise ValueError("No campaign lifetime_budget provided.") lifetime_budget_cents = int(campaign.lifetime_budget * 100) ad_campaign = await self.facebook_client.create_campaign( name=campaign.name, objective=campaign.objective.to_ad_campaign_objective( content_type=campaign.content_type ), lifetime_budget=lifetime_budget_cents, ) return ad_campaign async def create_engagement_audience_ad_set( self, campaign: Campaign, *, engagement_audience: EngagementAudience, campaign_external_id: str, ) -> CampaignAdSet: optimization_goal = campaign.objective.to_ad_set_optimization_goal( content_type=campaign.content_type ) args = CreateAdsetArgs( optimization_goal=optimization_goal, billing_event=AdSetBillingEvent.IMPRESSIONS, start_time=campaign.start_time, end_time=campaign.end_time, ) position = self._generate_ad_set_position(campaign) args.name = "engagement" audience = CustomAudience(id=engagement_audience.external_id) args.targeting = AdSetCustomAudienceTargeting( custom_audiences=[audience], publisher_platforms=position.publisher_platforms, facebook_positions=position.facebook_positions, instagram_positions=position.instagram_positions, ) engagement_audience_ad_set = await self.facebook_client.create_ad_set( campaign_id=campaign_external_id, args=args ) campaign_ad_set = CampaignAdSet() campaign_ad_set.external_id = engagement_audience_ad_set.id campaign_ad_set.campaign_id = campaign.id campaign_ad_set.name = args.name self.campaign_ad_set_repository.add(campaign_ad_set) campaign_ad_set_engagement_audience = CampaignAdSetEngagementAudience() campaign_ad_set_engagement_audience.ad_set_id = campaign_ad_set.id campaign_ad_set_engagement_audience.audience_id = engagement_audience.id self.campaign_ad_set_engagement_audience_repository.add( campaign_ad_set_engagement_audience ) return campaign_ad_set async def create_targeting_audience_ad_set( self, campaign: Campaign, targeting_audience: TargetingAudience, campaign_external_id: str, excluded_custom_audiences: list[CustomAudience], ) -> CampaignAdSet: optimization_goal = campaign.objective.to_ad_set_optimization_goal( content_type=campaign.content_type ) args = CreateAdsetArgs( optimization_goal=optimization_goal, billing_event=AdSetBillingEvent.IMPRESSIONS, start_time=campaign.start_time, end_time=campaign.end_time, ) position = self._generate_ad_set_position(campaign) args.name = "targeting" args.targeting = AdSetTargetingAudienceTargeting( geo_locations=targeting_audience.api_geo_locations, interests=targeting_audience.api_interests, age_min=targeting_audience.age_min, age_max=targeting_audience.age_max, publisher_platforms=position.publisher_platforms, facebook_positions=position.facebook_positions, instagram_positions=position.instagram_positions, excluded_custom_audiences=excluded_custom_audiences, ) targeting_audience_ad_set = await self.facebook_client.create_ad_set( campaign_id=campaign_external_id, args=args, ) campaign_ad_set = CampaignAdSet() campaign_ad_set.external_id = targeting_audience_ad_set.id campaign_ad_set.campaign_id = campaign.id campaign_ad_set.name = args.name self.campaign_ad_set_repository.add(campaign_ad_set) campaign_ad_set_targeting_audience = CampaignAdSetTargetingAudience() campaign_ad_set_targeting_audience.ad_set_id = campaign_ad_set.id campaign_ad_set_targeting_audience.audience_id = targeting_audience.id self.campaign_ad_set_targeting_audience_repository.add( campaign_ad_set_targeting_audience ) return campaign_ad_set async def create_ad_creative( self, campaign: Campaign, ) -> CampaignAdCreative: args: CreateAdCreativeArgs | None = None if not campaign.content_type == CampaignContentType.POST: raise ValueError( f"Unsupported campaign.content_type: {campaign.content_type}" ) else: if facebook_post_external_id := campaign.facebook_post_external_id: args = AdCreativeArgs(object_story_id=facebook_post_external_id) if campaign.instagram_user_external_id: args.instagram_actor_id = campaign.instagram_user_external_id if instagram_post_external_id := campaign.instagram_post_external_id: args = AdCreativeInstagramPostArgs( object_id=campaign.facebook_page.external_id, instagram_user_id=campaign.instagram_account_external_id, source_instagram_media_id=instagram_post_external_id, ) if not args: raise ValueError("No Ad Creative args created") ad_creative = await self.facebook_client.create_ad_creative(args) campaign_ad_creative = CampaignAdCreative() campaign_ad_creative.external_id = ad_creative.id campaign_ad_creative.campaign_id = campaign.id campaign_ad_creative.name = "test_name" self.campaign_ad_creative_repository.add(campaign_ad_creative) return campaign_ad_creative async def create_ad( self, campaign: Campaign, *, ad_set: CampaignAdSet, ad_creative: CampaignAdCreative, ) -> CampaignAd: ad = await self.facebook_client.create_ad( name=f"Ad - {ad_set.name}", ad_set_id=ad_set.external_id, ad_creative_id=ad_creative.external_id, ) campaign_ad = CampaignAd() campaign_ad.external_id = ad.id campaign_ad.campaign_id = campaign.id campaign_ad.ad_creative_id = ad_creative.id campaign_ad.ad_set_id = ad_set.id campaign_ad.name = "test_name" self.campaign_ad_repository.add(campaign_ad) return campaign_ad async def submit_ad_campaign(self, ads: list[CampaignAd]) -> None: for ad in ads: args = UpdateAdArgs(status=ObjectStatus.ACTIVE) await self.facebook_client.update_ad(ad_id=ad.external_id, args=args) def _generate_ad_set_position(self, campaign: Campaign) -> AdSetPosition: ad_formats = self.ad_format_service.get_campaign_available_ad_formats( campaign, content_type=campaign.content_type, platforms=campaign.platforms or [], ) facebook_positions = set() instagram_positions = set() for ad_format in ad_formats: if ad_format.platform == CampaignPlatform.FACEBOOK: facebook_positions.add(ad_format.position) if ad_format.platform == CampaignPlatform.INSTAGRAM: instagram_positions.add(ad_format.position) return AdSetPosition( publisher_platforms=campaign.platforms, facebook_positions=list(facebook_positions) if facebook_positions else None, instagram_positions=list(instagram_positions) if instagram_positions else None, )