import uuid from dataclasses import dataclass, field from marshmallow import validate, post_dump, post_load from datetime import date from typing import List, Optional, Union from handlers.enums import CampaignWorkflowStatus from shared.schemas import missing, CellStyleShema from media_plan.validators import CampaignExistsValidator, RequestApprovalValidator, SubmitApprovalValidator from campaigns.schemas import ( TerritoryResponseSchema, CampaignDetailsObjectiveResponseSchema, CampaignPlatformSchema, CampaignTypeSchema, CampaignDetailsAudienceAgeResponseSchema, CampaignDetailsGenderResponseSchema, CampaignDetailsProviderResponseSchema, CampaignStatusSchema, CampaignKpiMetricsSchema, CampaignPlacementSchema, ) from models.media_plans import MediaPlan from models.media_plan_phases import MediaPlanPhase from models.campaign import Campaign from models.campaign_approval import ApprovalRequest from api import ma import marshmallow_dataclass from flask_fsm import state_map from shared.validators import NotEmpty from users.schemas import UserModel from projects.schemas import TerritorySchema @dataclass class CreateUpdatePhaseRequest: """Create/Update project phase model""" order: Optional[int] = field(metadata={"validate": validate.Range(min=0)}, default=None) name: Optional[str] = field(default=None) with_campaign: Optional[bool] = field(default=True) campaign_uuid: Optional[str] = field(default=None, metadata={"data_key": "campaignUuid"}) create_update_phase_request_schema = marshmallow_dataclass.class_schema(CreateUpdatePhaseRequest, base_schema=ma.Schema) @dataclass class CreateUpdateMediaPlanRequest: name: Optional[str] = field(default=None, metadata={"validate": [validate.Length(min=1, max=40), NotEmpty()]}) # order: Optional[int] = field(metadata={"validate": validate.Range(min=0)}, default=None) create_update_media_plan_request_schema = marshmallow_dataclass.class_schema( CreateUpdateMediaPlanRequest, base_schema=ma.Schema ) @dataclass class CampaignStyleFieldsSchema: namingConventionManual: Optional[CellStyleShema] startDate: Optional[CellStyleShema] endDate: Optional[CellStyleShema] platforms: Optional[CellStyleShema] notes: Optional[CellStyleShema] creativeDescription: Optional[CellStyleShema] headline: Optional[CellStyleShema] callToAction: Optional[CellStyleShema] destinationLinks: Optional[CellStyleShema] audienceNotes: Optional[CellStyleShema] adCreativeNotes: Optional[CellStyleShema] kpiMetricsFieldId: Optional[CellStyleShema] objective: Optional[CellStyleShema] ecpm: Optional[CellStyleShema] name: Optional[CellStyleShema] namingConvention: Optional[CellStyleShema] actualSpent: Optional[CellStyleShema] plannedBudget: Optional[CellStyleShema] workflowStatus: Optional[CellStyleShema] genders: Optional[CellStyleShema] territories: Optional[CellStyleShema] audienceLowerAge: Optional[CellStyleShema] audienceHigherAge: Optional[CellStyleShema] placements: Optional[CellStyleShema] adCreativeLinks: Optional[CellStyleShema] estImpressions: Optional[CellStyleShema] @post_dump(pass_original=True) def post_dump_remove_none_values(self, data, obj, **kwargs): return {k: v for k, v in data.items() if v is not None} @post_load(pass_original=True) def remove_skip_values(self, data, original_data, **kwargs): return {k: v if k in original_data.keys() else "missing" for k, v in data.items()} @dataclass class MediaPlanCampaignColumnsSchema: id: int projectId: int placements: List[CampaignPlacementSchema] name: Optional[str] = field(default=None) namingConvention: Optional[str] = field(default=None) namingConventionManual: Optional[str] = field(default=None) plannedBudget: Optional[float] = field(default=None) budgetSpend: Optional[float] = field(default=None) ecpm: Optional[float] = field(default=None) startDate: Optional[date] = field(default=None) endDate: Optional[date] = field(default=None) notes: Optional[str] = field(default=None) orderInPhase: Optional[int] = field(default=None) actualSpent: Optional[float] = field(default=None) creativeDescription: Optional[str] = field(default=None) headline: Optional[str] = field(default=None) callToAction: Optional[str] = field(default=None) destinationLinks: Optional[List[str]] = field(default_factory=list) adCreativeLinks: Optional[List[str]] = field(default_factory=list) objective: Optional[CampaignDetailsObjectiveResponseSchema] = field(default=None) type: Optional[CampaignTypeSchema] = field(default=None) audienceAge: Optional[CampaignDetailsAudienceAgeResponseSchema] = field(default=None) kpiMetricsField: Optional[CampaignKpiMetricsSchema] = field(default=None) territories: Optional[List[TerritoryResponseSchema]] = field(default=None) platforms: Optional[List[CampaignPlatformSchema]] = field(default=None) provider: Optional[CampaignDetailsProviderResponseSchema] = field(default=None) genders: Optional[List[CampaignDetailsGenderResponseSchema]] = field(default=None) adCreativeNotes: Optional[str] = field(default=None) audienceNotes: Optional[str] = field(default=None) workflowStatus: Optional[CampaignStatusSchema] = field(default=None) uuid: Optional[str] = field(default=None) def __init__(self, campaign: Campaign): self.id = campaign.id self.projectId = campaign.project_id self.name = campaign.name self.namingConvention = campaign.naming_convention self.namingConventionManual = campaign.naming_convention_manual self.plannedBudget = campaign.planned_budget self.budgetSpend = campaign.budget_spend self.actualSpent = campaign.actual_spent self.startDate = campaign.start_date self.endDate = campaign.end_date self.orderInPhase = campaign.order_in_phase self.adCreativeNotes = campaign.ad_creative_notes self.audienceNotes = campaign.audience_notes self.notes = campaign.notes self.creativeDescription = campaign.creative_description self.headline = campaign.headline self.callToAction = campaign.call_to_action self.destinationLinks = [destination.link for destination in campaign.links] self.ecpm = campaign.ecpm if campaign.age_range: lower = campaign.age_range.lower if campaign.age_range and campaign.age_range.lower else None upper = campaign.age_range.upper - 1 if campaign.age_range and campaign.age_range.upper else None self.audienceAge = CampaignDetailsAudienceAgeResponseSchema(lower, upper) self.workflowStatus = CampaignStatusSchema( status=campaign.workflow_status, available_transitions=state_map.get_target(campaign.workflow_status) ) self.genders = [CampaignDetailsGenderResponseSchema(gender.id, gender.name) for gender in campaign.genders] self.platforms = [CampaignPlatformSchema(id=p.id, name=p.name, labelId=p.label_id) for p in campaign.platforms] self.territories = [TerritoryResponseSchema(at.territory.id, at.territory.name) for at in campaign.territories] self.adCreativeLinks = [link.url for link in campaign.ad_creative_links] self.placements = [ CampaignPlacementSchema(id=p.id, name=p.name, labelId=p.label_id) for p in campaign.placements ] self.uuid = campaign.uuid if campaign.kpi_metrics_field_id: self.kpiMetricsField = CampaignKpiMetricsSchema( id=campaign.kpi_metrics_field.id, name=campaign.kpi_metrics_field.name ) if campaign.objective: self.objective = CampaignDetailsObjectiveResponseSchema(campaign.objective.id, campaign.objective.name) if campaign.provider_id: self.provider = CampaignDetailsProviderResponseSchema(campaign.provider.id, campaign.provider.name) if campaign.type: self.type = CampaignTypeSchema(campaign.type) self.style = campaign.style @dataclass class MediaPlanCampaignSchema: columns: MediaPlanCampaignColumnsSchema style: Optional[CampaignStyleFieldsSchema] def __init__(self, campaign: Campaign): self.columns = MediaPlanCampaignColumnsSchema(campaign) self.style = campaign.style @dataclass class PhaseSchema: id: int name: str order: int media_plan_id: int def __init__(self, phase: MediaPlanPhase): self.id = phase.id self.name = phase.name self.order = phase.order self.media_plan_id = phase.media_plan_id phase_schema = marshmallow_dataclass.class_schema(PhaseSchema, base_schema=ma.Schema) @dataclass class MediaPlanPhaseSchema: id: int name: str order: int media_plan_id: int campaigns: List[MediaPlanCampaignSchema] def __init__(self, phase: MediaPlanPhase): self.id = phase.id self.name = phase.name self.order = phase.order self.media_plan_id = phase.media_plan_id self.campaigns = sorted([ MediaPlanCampaignSchema(campaign) for campaign in phase.campaigns if campaign.is_deleted is False ], key=lambda campaign: campaign.columns.orderInPhase) media_plan_phase_schema = marshmallow_dataclass.class_schema(MediaPlanPhaseSchema, base_schema=ma.Schema) @dataclass class MediaPlanResponseSchema: id: int name: str order: int def __init__(self, media_plan: MediaPlan): self.id = media_plan.id self.name = media_plan.name self.order = media_plan.order media_plan_response_schema = marshmallow_dataclass.class_schema(MediaPlanResponseSchema, base_schema=ma.Schema) @dataclass class MediaPlanShortResponseSchema: id: int name: str @dataclass class MediaPlanDetailsResponseSchema: phases: List[MediaPlanPhaseSchema] def __init__(self, media_plan: MediaPlan): self.phases = [MediaPlanPhaseSchema(phase) for phase in media_plan.phases] media_plan_details_response_schema = marshmallow_dataclass.class_schema( MediaPlanDetailsResponseSchema, base_schema=ma.Schema ) @dataclass class MediaPlanCampaignPartialUpdateSchema: naming_convention_manual: Optional[str] = field(default="missing", metadata={"data_key": "namingConventionManual"}) start_date: Optional[Union[date, missing]] = field(default="missing", metadata={"data_key": "startDate"}) end_date: Optional[Union[date, missing]] = field(default="missing", metadata={"data_key": "endDate"}) platforms: Optional[Union[List[int], missing]] = field(default="missing") notes: Optional[str] = field(default="missing") creative_description: Optional[str] = field(default="missing", metadata={"data_key": "creativeDescription"}) headline: Optional[str] = field(default="missing") call_to_action: Optional[str] = field(default="missing", metadata={"data_key": "callToAction"}) destination_links: Optional[List[str]] = field( default="missing", metadata={"data_key": "destinationLinks", "validate": [validate.Length(max=10)]} ) audience_notes: Optional[str] = field(default="missing", metadata={"data_key": "audienceNotes"}) ad_creative_notes: Optional[str] = field(default="missing", metadata={"data_key": "adCreativeNotes"}) kpi_metrics_field_id: Optional[Union[int, missing]] = field( default="missing", metadata={"data_key": "kpiMetricsFieldId"} ) objective_id: Optional[Union[int, missing]] = field(default="missing", metadata={"data_key": "objective"}) ecpm: Optional[Union[float, missing]] = field(default="missing") name: Optional[str] = field(default="missing") naming_convention: Optional[str] = field(default="missing", metadata={"data_key": "namingConvention"}) actual_spent: Optional[Union[missing, float]] = field(default="missing", metadata={"data_key": "actualSpent"}) planned_budget: Optional[Union[float, missing]] = field(default="missing", metadata={"data_key": "plannedBudget"}) workflow_status: Optional[Union[str, missing]] = field(default="missing", metadata={"data_key": "workflowStatus"}) genders: Optional[Union[List[int], missing]] = field(default="missing") territories: Optional[Union[List[int], missing]] = field(default="missing") audience_lower_age: Optional[Union[int, missing]] = field( default="missing", metadata={"data_key": "audienceLowerAge"} ) audience_higher_age: Optional[Union[int, missing]] = field( default="missing", metadata={"data_key": "audienceHigherAge"} ) placements: Optional[Union[List[int], missing]] = field(default="missing") ad_creative_links: Optional[Union[List[Optional[str]], missing]] = field( default="missing", metadata={"data_key": "adCreativeLinks"} ) media_plan_campaign_partial_update_schema = marshmallow_dataclass.class_schema(MediaPlanCampaignPartialUpdateSchema, base_schema=ma.Schema) @dataclass class MediaPlanCampaignPartialUpdateResponseSchema: namingConvention: str media_plan_campaign_partial_update_response_schema = marshmallow_dataclass.class_schema( MediaPlanCampaignPartialUpdateResponseSchema, base_schema=ma.Schema ) @dataclass class MediaPlanCampaignsBulkActionUpdateCampaignSchema: columns: MediaPlanCampaignPartialUpdateSchema style: Optional[Union[CampaignStyleFieldsSchema, missing]] = field(default="missing") campaign_uuid: str = field(default=None, metadata={"data_key": "campaignUuid"}) phase_id: Optional[int] = field(default=None, metadata={"data_key": "phaseId"}) order_in_phase: Optional[int] = field(default=None, metadata={"data_key": "orderInPhase"}) @dataclass class MediaPlanCampaignsBulkActionCreateCampaignSchema: columns: MediaPlanCampaignPartialUpdateSchema style: Optional[Union[CampaignStyleFieldsSchema, missing]] = field(default="missing") phase_id: int = field(default=None, metadata={"data_key": "phaseId"}) order_in_phase: int = field(default=None, metadata={"data_key": "orderInPhase"}) campaign_uuid: str = field(default=None, metadata={"data_key": "campaignUuid"}) @dataclass class MediaPlanCampaignsBulkActionSchema: update: Optional[List[MediaPlanCampaignsBulkActionUpdateCampaignSchema]] = field(default_factory=list) create: Optional[List[MediaPlanCampaignsBulkActionCreateCampaignSchema]] = field(default_factory=list) delete: Optional[List[str]] = field(default_factory=list) media_plan_campaign_bulk_action_schema = marshmallow_dataclass.class_schema( MediaPlanCampaignsBulkActionSchema, base_schema=ma.Schema ) @dataclass class MediaPlanCampaignsBulkActionCampaignResponseSchema: id: int namingConvention: str uuid: Optional[str] = field(default=None) @dataclass class MediaPlanCampaignsBulkActionResponseSchema: created: List[MediaPlanCampaignsBulkActionCampaignResponseSchema] updated: List[MediaPlanCampaignsBulkActionCampaignResponseSchema] media_plan_campaign_bulk_action_response_schema = marshmallow_dataclass.class_schema( MediaPlanCampaignsBulkActionResponseSchema, base_schema=ma.Schema ) @dataclass class ApproverUserResponseSchema: id: int name: str email: str hasProjectAccess: bool approver_user_response_schema = marshmallow_dataclass.class_schema( ApproverUserResponseSchema, base_schema=ma.Schema ) @dataclass class ApprovalRequestSchema: campaigns: List[uuid.UUID] = field(metadata={"validate": [ validate.Length(min=1), CampaignExistsValidator(), RequestApprovalValidator() ]}) reviewers: List[int] = field(metadata={"validate": validate.Length(min=1, max=1)}) message: Optional[str] approval_request_schema = marshmallow_dataclass.class_schema( ApprovalRequestSchema, base_schema=ma.Schema ) @dataclass class ApprovalRequestResponseSchema: approvalRequestId: int def __init__(self, request: ApprovalRequest): self.approvalRequestId = request.id approval_request_response_schema = marshmallow_dataclass.class_schema( ApprovalRequestResponseSchema, base_schema=ma.Schema ) @dataclass class MediaPlanReviewPhaseSchema: id: int name: str order: int media_plan_id: int campaigns: List[MediaPlanCampaignSchema] def __init__(self, phase: MediaPlanPhase, campaigns): self.id = phase.id self.name = phase.name self.order = phase.order self.media_plan_id = phase.media_plan_id self.campaigns = sorted( [MediaPlanCampaignSchema(campaign) for campaign in campaigns if campaign.is_deleted is False], key=lambda campaign: campaign.columns.orderInPhase ) @dataclass class MediaPlanReviewResponseSchema: phases: List[MediaPlanReviewPhaseSchema] requester: UserModel def __init__(self, phases, requester): self.phases = phases self.requester = UserModel(requester) media_plan_review_response_schema = marshmallow_dataclass.class_schema( MediaPlanReviewResponseSchema, base_schema=ma.Schema ) @dataclass class SubmitApprovalCampaignSchema: campaign_uuid: str = field(metadata={"data_key": "campaignUuid", "validate": [ CampaignExistsValidator(), SubmitApprovalValidator() ]}) workflow_status: str = field(metadata={ "data_key": "workflowStatus", "validate": validate.OneOf([CampaignWorkflowStatus.APPROVED.value, CampaignWorkflowStatus.DISAPPROVED.value]) }) @dataclass class SubmitApprovalSchema: campaigns: List[SubmitApprovalCampaignSchema] submit_approval_schema = marshmallow_dataclass.class_schema( SubmitApprovalSchema, base_schema=ma.Schema ) @dataclass class MediaPlanMetricsResponseSchema: campaignsCount: int platformsCount: int estimatedImpressions: float plannedBudget: float territories: List[TerritorySchema] media_plan_metrics_response_schema = marshmallow_dataclass.class_schema( MediaPlanMetricsResponseSchema, base_schema=ma.Schema ) @dataclass class MediaPlanReleaseDetails: name: str type: int media_plan_release_details = marshmallow_dataclass.class_schema( MediaPlanReleaseDetails, base_schema=ma.Schema ) @dataclass class MediaPlanReleaseDetailsUpdate: name: Optional[Union[str, missing]] = field(default="missing") type: Optional[Union[int, missing]] = field(default="missing") media_plan_release_details_update = marshmallow_dataclass.class_schema( MediaPlanReleaseDetailsUpdate, base_schema=ma.Schema ) @dataclass class CampaignNamingConventionResponse: campaignUuid: uuid.UUID namingConvention: str def __init__(self, campaign_uuid, name): self.campaignUuid = campaign_uuid self.namingConvention = name @dataclass class MediaPlanReleaseDetailsUpdateResponse: name: Optional[str] type: Optional[int] campaignsNames: List[CampaignNamingConventionResponse] media_plan_release_details_update_response = marshmallow_dataclass.class_schema( MediaPlanReleaseDetailsUpdateResponse, base_schema=ma.Schema )