import datetime from collections.abc import Callable from typing import Any, TypeVar, cast import factory.fuzzy from audience_common import datetimeutc from audience_common.types import UTCDateTime from factory.alchemy import SQLAlchemyModelFactory from polyfactory.factories.pydantic_factory import ModelFactory as BaseModelFactory from pydantic import BaseModel from campaigns.assets.enums import AssetStatus from campaigns.assets.models import Asset from campaigns.campaigns.enums import CampaignStatus from campaigns.core.types import EncryptedToken from campaigns.meta.enums import ( AssetPositionGroup, AudienceGender, CampaignContentType, CampaignObjective, CampaignPlatform, ) from campaigns.meta.models import ( Campaign, CampaignAd, CampaignAdCreative, CampaignAdSet, CampaignImage, CampaignVideo, EngagementAudience, FacebookAssociation, FacebookPage, FacebookPageLabel, TargetingAudience, ) T = TypeVar("T", bound=BaseModel) class ModelFactory(BaseModelFactory[T]): __allow_none_optionals__ = False __is_base_factory__ = True @classmethod def get_provider_map(cls) -> dict[Any, Callable[..., Any]]: result = super().get_provider_map() result[UTCDateTime] = lambda: cast( UTCDateTime, cls.__faker__.date_time_between(tzinfo=datetime.UTC) ) return result # Assets class AssetFactory(SQLAlchemyModelFactory): id = factory.Faker("uuid4") status = AssetStatus.COMPLETED filename: str = "asset.jpg" key: str = factory.LazyAttribute(lambda asset: f"raw/{asset.id}/{asset.filename}") bucket: str = "assets-bucket" filetype: str = "image" content_type: str = "image/jpeg" thumbnail_key = factory.LazyAttribute( lambda asset: f"thumbnails/{asset.id}/{asset.filename}{'.jpg' if asset.filetype == 'video' else ''}" ) class Meta: model = Asset # Meta class FacebookAssociationFactory(SQLAlchemyModelFactory): user_id = factory.Faker("pystr_format", string_format="############{{random_int}}") user_name = factory.Faker("name") orchard_identity_id = factory.Faker("uuid4") token = factory.LazyFunction(lambda: EncryptedToken("token")) expires_at = factory.LazyFunction( lambda: datetimeutc.now() + datetime.timedelta(days=60) ) class Meta: model = FacebookAssociation sqlalchemy_get_or_create = ("orchard_identity_id",) class FacebookPageFactory(SQLAlchemyModelFactory): association = factory.SubFactory(FacebookAssociationFactory) external_id = factory.Faker( "pystr_format", string_format="###########{{random_int}}" ) name = factory.Faker("pystr") token = factory.LazyFunction(lambda: EncryptedToken("token")) is_manageable = True class Meta: model = FacebookPage sqlalchemy_get_or_create = ("association", "external_id") @classmethod def _adjust_kwargs(cls, **kwargs: Any) -> Any: if "association" in kwargs and "association_id" not in kwargs: kwargs["association_id"] = kwargs["association"].id return kwargs class FacebookPageLabelFactory(SQLAlchemyModelFactory): page = factory.SubFactory(FacebookPageFactory) vendor_id = factory.Faker("pyint", min_value=1) subaccount_id = factory.Faker("pyint", min_value=0) class Meta: model = FacebookPageLabel sqlalchemy_get_or_create = ("page", "vendor_id", "subaccount_id") @classmethod def _adjust_kwargs(cls, **kwargs: Any) -> Any: if "page" in kwargs and "page_id" not in kwargs: kwargs["page_id"] = kwargs["page"].id return kwargs class MetaTargetingAudienceFactory(SQLAlchemyModelFactory): locations: list[dict[str, Any]] = [] age_min = 18 age_max = 65 genders = [AudienceGender.MALE, AudienceGender.FEMALE] interests: list[dict[str, Any]] = [] class Meta: model = TargetingAudience class MetaEngagementAudienceFactory(SQLAlchemyModelFactory): class Meta: model = EngagementAudience class MetaCampaignAdSetFactory(SQLAlchemyModelFactory): class Meta: model = CampaignAdSet class MetaCampaignAdCreativeFactory(SQLAlchemyModelFactory): class Meta: model = CampaignAdCreative class MetaCampaignAdFactory(SQLAlchemyModelFactory): class Meta: model = CampaignAd class MetaCampaignFactory(SQLAlchemyModelFactory): name = "Test campaign" vendor_id = factory.Faker("pyint", min_value=1) subaccount_id = factory.Faker("pyint", min_value=0) global_participant_id = factory.Faker("uuid4") status = CampaignStatus.DRAFT objective = CampaignObjective.TRAFFIC content_type = CampaignContentType.IMAGE start_time = factory.Faker("past_datetime", tzinfo=datetime.UTC) end_time = factory.Faker("future_datetime", tzinfo=datetime.UTC) lifetime_budget = factory.Faker( "pydecimal", positive=True, max_value=10000, right_digits=2 ) platforms = [CampaignPlatform.FACEBOOK, CampaignPlatform.INSTAGRAM] created_by = factory.Faker("uuid4") updated_by = factory.Faker("uuid4") # Relations facebook_page = factory.SubFactory(FacebookPageFactory) targeting_audience = factory.SubFactory(MetaTargetingAudienceFactory) images: list[Any] = [] videos: list[Any] = [] class Meta: model = Campaign @classmethod def _adjust_kwargs(cls, **kwargs: Any) -> Any: if "facebook_page" in kwargs and "facebook_page_id" not in kwargs: kwargs["facebook_page_id"] = kwargs["facebook_page"].id if ( targeting_audience := kwargs.get("targeting_audience") ) and "targeting_audience_id" not in kwargs: kwargs["targeting_audience_id"] = targeting_audience.id return kwargs class MetaCampaignImageFactory(SQLAlchemyModelFactory): campaign = factory.SubFactory(MetaCampaignFactory) asset = factory.SubFactory(AssetFactory) position_group = AssetPositionGroup.RATIO_1_1 class Meta: model = CampaignImage @classmethod def _adjust_kwargs(cls, **kwargs: Any) -> Any: if "asset" in kwargs and "asset_id" not in kwargs: kwargs["asset_id"] = kwargs["asset"].id return kwargs class MetaCampaignVideoFactory(SQLAlchemyModelFactory): campaign = factory.SubFactory(MetaCampaignFactory) asset = factory.SubFactory(AssetFactory) position_group = AssetPositionGroup.RATIO_1_1 class Meta: model = CampaignVideo @classmethod def _adjust_kwargs(cls, **kwargs: Any) -> Any: if "asset" in kwargs and "asset_id" not in kwargs: kwargs["asset_id"] = kwargs["asset"].id return kwargs