from dataclasses import dataclass from campaigns.auth.dtos import User from campaigns.connectors.db import Database, session_context from campaigns.connectors.facebook.arguments import ( AdCreativeArgs, AdCreativeInstagramPostArgs, GetAdPreviewArgs, ) from campaigns.connectors.facebook.base import FacebookClient from campaigns.connectors.facebook.enums import AdCreativeCallToActionType from campaigns.connectors.facebook.models import ( AdCreativeCallToAction, AdCreativeCallToActionValue, AdCreativeImageStorySpec, AdCreativeLinkData, AdCreativeVideoData, AdCreativeVideoStorySpec, ) from campaigns.core.handler import Handler from campaigns.meta import constants from campaigns.meta.dtos import AdFormat from campaigns.meta.enums import CampaignCallToAction, CampaignContentType from campaigns.meta.exceptions import ( CampaignAdFormatDoesNotExist, CampaignAdImageDoesNotExist, CampaignAdPreviewIsUnavailable, CampaignAdVideoDoesNotExist, ) from campaigns.meta.models import Campaign from campaigns.meta.repositories import AdFormatRepository from campaigns.meta.services import CampaignPostService, CampaignService @dataclass(frozen=True) class GetCampaignAdPreviewRequest: user: User campaign_id: str ad_format_id: str content_type: CampaignContentType post_external_id: str | None primary_text: str | None headline: str | None description: str | None website_url: str | None call_to_action: CampaignCallToAction | None def __post_init__(self) -> None: if self.primary_text is None: object.__setattr__( self, "primary_text", constants.DEFAULT_CAMPAIGN_AD_PRIMARY_TEXT ) if self.website_url is None: object.__setattr__( self, "website_url", constants.DEFAULT_CAMPAIGN_AD_WEBSITE_URL ) if self.call_to_action is None: object.__setattr__( self, "call_to_action", constants.DEFAULT_CAMPAIGN_AD_CALL_TO_ACTION ) @dataclass(frozen=True) class GetCampaignAdPreviewResponse: body: str src: str height: int width: int @dataclass class GetCampaignAdPreviewHandler( Handler[GetCampaignAdPreviewRequest, GetCampaignAdPreviewResponse], ): db: Database facebook_client: FacebookClient campaign_service: CampaignService ad_format_repository: AdFormatRepository campaign_post_service: CampaignPostService @session_context(in_transaction=False) async def handle( self, request: GetCampaignAdPreviewRequest ) -> GetCampaignAdPreviewResponse: campaign = await self.campaign_service.get_user_campaign( user=request.user, campaign_id=request.campaign_id ) ad_format = self.ad_format_repository.get_or_none(request.ad_format_id) if not ad_format: raise CampaignAdFormatDoesNotExist call_to_action_type = AdCreativeCallToActionType( ( request.call_to_action or constants.DEFAULT_CAMPAIGN_AD_CALL_TO_ACTION ).value.upper() ) call_to_action = AdCreativeCallToAction( type=AdCreativeCallToActionType(call_to_action_type), value=AdCreativeCallToActionValue(link=request.website_url), ) creative: AdCreativeArgs | AdCreativeInstagramPostArgs if request.content_type == CampaignContentType.POST: post = await self.campaign_post_service.get_campaign_post( campaign, post_external_id=request.post_external_id ) if post.is_instagram: creative = AdCreativeInstagramPostArgs( object_id=campaign.facebook_page.external_id, instagram_user_id=campaign.instagram_account_external_id, source_instagram_media_id=post.id, ) else: creative = AdCreativeArgs( object_story_id=post.id, instagram_actor_id=campaign.instagram_user_external_id, ) else: if request.content_type == CampaignContentType.VIDEO: creative = await self._get_video_creative( campaign=campaign, ad_format=ad_format, call_to_action=call_to_action, request=request, ) elif request.content_type == CampaignContentType.IMAGE: creative = self._get_image_creative( campaign=campaign, ad_format=ad_format, call_to_action=call_to_action, request=request, ) else: raise TypeError( f"Unsupported campaign content type: {request.content_type.value}" ) ad_preview = await self.facebook_client.get_ad_preview( GetAdPreviewArgs(ad_format=ad_format.id, creative=creative), ) if not ad_preview.src: raise CampaignAdPreviewIsUnavailable return GetCampaignAdPreviewResponse( body=ad_preview.body, src=ad_preview.src, height=ad_format.height, width=ad_format.width, ) async def _get_video_creative( self, campaign: Campaign, ad_format: AdFormat, call_to_action: AdCreativeCallToAction, request: GetCampaignAdPreviewRequest, ) -> AdCreativeArgs: videos = [ video for video in campaign.videos if video.position_group == ad_format.position_group ] if videos and videos[0].external_id: ad_video = await self.facebook_client.get_ad_video( video_id=videos[0].external_id ) else: raise CampaignAdVideoDoesNotExist return AdCreativeArgs( object_story_spec=AdCreativeVideoStorySpec( page_id=campaign.facebook_page.external_id, instagram_actor_id=campaign.instagram_user_external_id, video_data=AdCreativeVideoData( title=request.headline, video_id=ad_video.id, image_url=ad_video.thumbnail_url, call_to_action=call_to_action, message=request.primary_text, ), ) ) @staticmethod def _get_image_creative( campaign: Campaign, ad_format: AdFormat, call_to_action: AdCreativeCallToAction, request: GetCampaignAdPreviewRequest, ) -> AdCreativeArgs: images = [ image for image in campaign.images if image.position_group == ad_format.position_group ] if images: image = images[0] else: raise CampaignAdImageDoesNotExist return AdCreativeArgs( object_story_spec=AdCreativeImageStorySpec( page_id=campaign.facebook_page.external_id, instagram_actor_id=campaign.instagram_user_external_id, link_data=AdCreativeLinkData( name=request.headline, description=request.description, message=request.primary_text, link=request.website_url, image_hash=image.image_hash, call_to_action=call_to_action, ), ) )