from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from campaigns.connectors.facebook import models as fb_models from campaigns.connectors.facebook.enums import ( AdSetFacebookPosition, AdSetInstagramPosition, AudienceEventSourceType, EngagementAudienceEventValue, ) from campaigns.core.exceptions import FieldError from campaigns.meta import constants from campaigns.meta.enums import ( AdFormatGroup, AssetPositionGroup, CampaignPlatform, CampaignSummaryId, FacebookConnectionStatus, ) @dataclass class FacebookUser: id: str name: str picture: str | None = None granted_scopes: list[str] = field(default_factory=list) def scopes_sufficient( self, *, scopes: list[str] = constants.REQUIRED_SCOPES ) -> bool: return set(self.granted_scopes).issuperset(scopes) def has_pages_scope(self) -> bool: return constants.SCOPE_PAGES_SHOW_LIST in self.granted_scopes @classmethod def from_api_obj( cls, obj: fb_models.User, granted_scopes: list[str] | None = None ) -> FacebookUser: return cls( id=obj.id, name=obj.name, picture=obj.picture.data.url, granted_scopes=granted_scopes or [], ) @dataclass class FacebookConnection: user: FacebookUser | None status: FacebookConnectionStatus scopes: list[str] = field(default_factory=lambda: constants.REQUEST_SCOPES) # TODO: deprecated required_scopes: list[str] = field( default_factory=lambda: constants.REQUIRED_SCOPES ) @property def is_connected(self) -> bool: return self.status == FacebookConnectionStatus.CONNECTED @classmethod def not_connected(cls) -> FacebookConnection: return cls( user=None, status=FacebookConnectionStatus.NOT_CONNECTED, ) @dataclass class InstagramAccount: id: str name: str | None username: str profile_picture_url: str | None @classmethod def from_api_obj(cls, obj: fb_models.InstagramBusinessAccount) -> InstagramAccount: return InstagramAccount( id=obj.id, name=obj.name, username=obj.username, profile_picture_url=obj.profile_picture_url, ) @dataclass(frozen=True) class CampaignPost: id: str platform: CampaignPlatform type: str title: str | None permalink_url: str picture_url: str | None thumbnail_url: str | None created_at: datetime @property def is_instagram(self) -> bool: return self.platform == CampaignPlatform.INSTAGRAM @classmethod def from_api_obj( cls, obj: fb_models.InstagramMedia | fb_models.PagePost ) -> CampaignPost: if isinstance(obj, fb_models.InstagramMedia): return cls( id=obj.id, platform=CampaignPlatform.INSTAGRAM, type=obj.media_product_type.lower().title(), title=obj.caption, permalink_url=obj.permalink, picture_url=obj.media_url, thumbnail_url=obj.thumbnail_url, created_at=obj.timestamp, ) return cls( id=obj.id, platform=CampaignPlatform.FACEBOOK, type="Post", title=obj.message, permalink_url=obj.permalink_url, picture_url=obj.full_picture, thumbnail_url=obj.picture, created_at=obj.created_time, ) @dataclass(frozen=True) class AdFormat: id: str name: str position: AdSetFacebookPosition | AdSetInstagramPosition group: AdFormatGroup position_group: AssetPositionGroup platform: CampaignPlatform height: int width: int @dataclass(frozen=True) class EngagementAudienceEvent: id: str name: str platform: CampaignPlatform source_type: AudienceEventSourceType value: EngagementAudienceEventValue @dataclass() class EngagementAudienceKey: facebook_page_external_id: str | None instagram_account_external_id: str | None event_ids: list[str] @dataclass(frozen=True) class CampaignValidationSummary: id: CampaignSummaryId errors: list[FieldError]