from dataclasses import dataclass from campaigns.auth.dtos import User from campaigns.connectors.db import Database, session_context from campaigns.connectors.facebook.arguments import GetAudienceSummaryArgs from campaigns.connectors.facebook.base import FacebookClient from campaigns.connectors.facebook.enums import ( AdSetFacebookPosition, AdSetInstagramPosition, ) from campaigns.connectors.facebook.models import ( AudienceSummary, TargetingGeolocations, TargetingInterestShort, ) from campaigns.core.handler import Handler from campaigns.meta.enums import CampaignPlatform from campaigns.meta.services import AdFormatService, CampaignService @dataclass(frozen=True) class GetCampaignAudienceSummaryRequest: user: User campaign_id: str geo_locations: TargetingGeolocations | None = None interests: list[TargetingInterestShort] | None = None genders: list[int] | None = None age_min: int | None = None age_max: int | None = None @dataclass class GetCampaignAudienceSummaryHandler( Handler[GetCampaignAudienceSummaryRequest, AudienceSummary], ): db: Database facebook_client: FacebookClient campaign_service: CampaignService ad_format_service: AdFormatService @session_context(in_transaction=False) async def handle( self, request: GetCampaignAudienceSummaryRequest ) -> AudienceSummary: campaign = await self.campaign_service.get_user_campaign( user=request.user, campaign_id=request.campaign_id ) get_summary_args = GetAudienceSummaryArgs( geo_locations=request.geo_locations, interests=request.interests, genders=request.genders, age_min=request.age_min, age_max=request.age_max, ) if campaign.platforms: get_summary_args.publisher_platforms = [ platform.to_ad_campaign_platform() for platform in campaign.platforms ] if campaign.content_type: ad_formats = self.ad_format_service.get_campaign_available_ad_formats( campaign, content_type=campaign.content_type, platforms=campaign.platforms or [], ) facebook_positions: list[AdSetFacebookPosition] = [] instagram_positions: list[AdSetInstagramPosition] = [] for ad_format in ad_formats: if ad_format.platform == CampaignPlatform.FACEBOOK: facebook_positions.append( AdSetFacebookPosition(ad_format.position.value) ) elif ad_format.platform == CampaignPlatform.INSTAGRAM: instagram_positions.append( AdSetInstagramPosition(ad_format.position.value) ) else: raise TypeError( f"Not supported Campaign Platform " f"type: {ad_format.platform}" ) get_summary_args.facebook_positions = facebook_positions or None get_summary_args.instagram_positions = instagram_positions or None return await self.facebook_client.get_audience_summary(get_summary_args)