"""Schemas for projects and campaigns.""" from datetime import date, datetime from enum import Enum import marshmallow_dataclass from api import ma from config import TIMEZONE, LABEL_IMAGE_SERVICE_URL from dataclasses import dataclass, field from typing import Optional, List, Any from marshmallow import validate, ValidationError, fields from models.user_project import UserProjectRoles from projects.constants import ProjectTargetTypes, MAX_RECENTLY_VIEWED_PROJECTS, MAX_NEWLY_ADDED_PROJECT from constants.project_statuses import ProjectStatuses from constants.pagination import PAGE_LIMIT_DEFAULT, PAGE_OFFSET_DEFAULT from shared.schemas import ArtistImageSchema, GRASDataSchema, PRSDataSchema, CCPDataSchema from projects.models import ProjectCalendarItemModel, ProjectSpendByDayModel, UsersSearchModel from utils.handlers import to_schema from models.product_track import ProductTrack from models.projects import ProjectSource, Project from playlists.schemas import PlaylistResponseSchema from marshmallow_dataclass import NewType from shared.validators import NotEmpty, UniqueValidator from shared.schemas import ( LinkfireLinkResponseSchema, ProjectLinkfireLinkListMapper, CampaignPlatformsMetadata, RoleSchema, LabelSchema, LabelResponseModel ) class ProjectArtistType(Enum): PRIMARY = "Primary" FEATURED = "Featured" @staticmethod def all_values() -> List[str]: # noinspection PyTypeChecker return list(map(lambda c: c.value, ProjectArtistType)) class ProjectTargetsListField(fields.Field): def _serialize(self, value: List, attr, obj, **kwargs): target_type = ProjectTargetTypes.from_value(obj.type) if target_type == ProjectTargetTypes.ARTIST: return to_schema(ProjectArtistSchema, many=True).dump(value) if target_type == ProjectTargetTypes.PLAYLIST: return to_schema(PlaylistResponseSchema, many=True).dump(value) return [] def _deserialize(self, value, attr, data, **kwargs): self._validate(value) target_type = ProjectTargetTypes.from_value(data["type"]) if target_type == ProjectTargetTypes.ARTIST: return to_schema(ProjectArtistSchema, many=True).load(value) if target_type == ProjectTargetTypes.PLAYLIST: return to_schema(PlaylistResponseSchema, many=True).load(value) return [] TerritoryID = NewType("TerritoryID", int) ProjectID = NewType("ProjectID", int, validate=validate.Range(min=1)) CampaignTypeID = NewType("CampaignTypeID", int, validate=validate.Range(min=1)) ProjectStatus = NewType("ProjectStatus", int, validate=validate.OneOf([e.value for e in ProjectStatuses])) ProjectTargets = NewType("ProjectTargets", list, validate=validate.Length(min=1), field=ProjectTargetsListField) class PhoneNumber(validate.Validator): def __call__(self, value) -> Any: message = "Phone number must be empty or at least 9 digits" if value and len(value) < 9: raise ValidationError(message) return value @dataclass class ProjectTargetSchema: type: str = field(metadata={"validate": validate.OneOf(ProjectTargetTypes.all_values())}) items: ProjectTargets class PrimaryArtistsValidator(validate.Validator): def __call__(self, value: ProjectTargetSchema) -> Any: if value.type != ProjectTargetTypes.ARTIST.value: return value primary_artists = list(filter(lambda x: x.type == ProjectArtistType.PRIMARY.value, value.items)) if len(primary_artists) > 3: raise ValidationError("You can select up to 3 primary artists.") if len(primary_artists) == 0: raise ValidationError("You must select at least one primary artist.") return value class FeaturedArtistsValidator(validate.Validator): def __call__(self, value: ProjectTargetSchema) -> Any: if value.type != ProjectTargetTypes.ARTIST.value: return value featured_artists = list(filter(lambda x: x.type == ProjectArtistType.FEATURED.value, value.items)) if len(featured_artists) > 20: raise ValidationError("You can select up to 20 featured artists.") return value class PlaylistsValidator(validate.Validator): def __call__(self, value: ProjectTargetSchema) -> Any: if value.type != ProjectTargetTypes.PLAYLIST.value: return value if len(value.items) > 20: raise ValidationError("You can select up to 20 playlists.") return value @dataclass class ProjectListQueryParams: statuses: Optional[List[ProjectStatus]] earliestStartDate: Optional[date] latestEndDate: Optional[date] freeText: Optional[str] = field(metadata={"validate": validate.Length(min=1, max=500)}) limit: Optional[int] = field(metadata={"validate": validate.Range(min=1)}) type: Optional[str] = field(metadata={"validate": validate.OneOf(ProjectTargetTypes.all_values())}) sort: Optional[str] = field(default="status") onlyMine: Optional[bool] = field(default=False) labels: Optional[List[int]] = field(default_factory=list) offset: Optional[int] = field(default=PAGE_OFFSET_DEFAULT, metadata={"validate": validate.Range(min=0)}) timeZone: Optional[str] = field(default=TIMEZONE) isClaimed: Optional[bool] = field(default=True) project_list_query_params = marshmallow_dataclass.class_schema( ProjectListQueryParams, base_schema=ma.Schema ) @dataclass class UserSchema: id: str email: str def __init__(self, user): self.id = user.external_id self.email = user.email @dataclass class ProjectModelSchema: id: int name: str title: str targets: ProjectTargetSchema budget: float allocation: float totalSpend: float startDate: Optional[date] = field(metadata={"format": "%Y-%m-%d"}) endDate: Optional[date] = field(metadata={"format": "%Y-%m-%d"}) status: int sharedStatus: Optional[str] userRoles: List[RoleSchema] isConfidential: bool currency: str isClaimed: bool @dataclass class RecentProjectModelSchema(ProjectModelSchema): campaignsCount: int @dataclass class BaseCountModel: artistCount: int playlistCount: int totalCount: int @dataclass class ProjectsCountResponseSchema: inProgress: BaseCountModel scheduled: BaseCountModel completed: BaseCountModel unassigned: BaseCountModel total: BaseCountModel def __init__(self, metadata): self.inProgress = BaseCountModel( artistCount=metadata.in_progress_artist, playlistCount=metadata.in_progress_playlist, totalCount=metadata.in_progress_total, ) self.completed = BaseCountModel( artistCount=metadata.completed_artist, playlistCount=metadata.completed_playlist, totalCount=metadata.completed_total, ) self.scheduled = BaseCountModel( artistCount=metadata.scheduled_artist, playlistCount=metadata.scheduled_playlist, totalCount=metadata.scheduled_total, ) self.unassigned = BaseCountModel( artistCount=metadata.unassigned_artists, playlistCount=0, totalCount=metadata.unassigned_artists ) self.total = BaseCountModel( artistCount=metadata.total_artists + metadata.unassigned_total, playlistCount=metadata.total_playlists, totalCount=metadata.total + metadata.unassigned_total, ) @dataclass class ProjectListResponseSchema: metadata: ProjectsCountResponseSchema items: List[ProjectModelSchema] = field(default_factory=list) limit: int = field(default_factory=int) offset: int = field(default_factory=int) project_list_response_schema = marshmallow_dataclass.class_schema( ProjectListResponseSchema, base_schema=ma.Schema ) @dataclass class RecordParams: typeId: int name: Optional[str] @dataclass class ContactParams: userId: int name: str email: str = field(metadata={"validate": validate.Email()}) phone: str = field(metadata={"validate": PhoneNumber()}) @dataclass class UpdatingEntityParams: entityId: int startDate: date endDate: Optional[date] @dataclass class ProjectScheduleUpdateParams: phases: Optional[List[UpdatingEntityParams]] campaigns: Optional[List[UpdatingEntityParams]] @dataclass class UpdateProjectParams: name: str = field(metadata={"validate": NotEmpty()}) targets: ProjectTargetSchema = field( metadata={"validate": [PrimaryArtistsValidator(), FeaturedArtistsValidator(), PlaylistsValidator()]} ) notes: Optional[str] startDate: date endDate: date scheduleUpdates: Optional[ProjectScheduleUpdateParams] linkfireLinks: Optional[List[LinkfireLinkResponseSchema]] = field(default_factory=list) update_project_params = marshmallow_dataclass.class_schema(UpdateProjectParams, base_schema=ma.Schema) @dataclass class ContactSchema: id: int externalId: str name: str email: str phone: str def __init__(self, contact): self.id = contact.id self.externalId = contact.external_id self.name = contact.name self.email = contact.email self.phone = contact.phone @dataclass class TerritorySchema: id: int name: str def __init__(self, territory): self.id = territory.territory_id self.name = territory.name @dataclass class CountByCategoryModel: id: int name: str count: int @dataclass class TeamCountersResponseSchema: usersCount: int team_counters_response_schema = marshmallow_dataclass.class_schema( TeamCountersResponseSchema, base_schema=ma.Schema ) @dataclass class ProjectBudgetDataResponseSchema: budgetData: List[ProjectSpendByDayModel] @dataclass class ProjectLinkfireMetadata: projectLinksCount: int campaignsLinksCount: int @dataclass class ProjectCampaignsMetadata: internalCampaignsCount: int externalCampaignsCount: int pendingCampaignsCount: int @dataclass class ProjectMetadata: campaigns: ProjectCampaignsMetadata productFamiliesCount: int assignedUsersCount: int linkfire: ProjectLinkfireMetadata estimatedImpressions: float campaignPlatforms: List[CampaignPlatformsMetadata] plannedBudget: int earliestStartDate: Optional[date] latestEndDate: Optional[date] @dataclass class ProjectResponseSchema: id: int name: str budget: float allocation: float editDate: datetime totalSpend: int status: int label: LabelResponseModel editUser: UserSchema createUser: UserSchema targets: ProjectTargetSchema territories: List[TerritorySchema] sharedStatus: Optional[str] userRoles: List[RoleSchema] notes: Optional[str] startDate: date endDate: date grasData: GRASDataSchema prsData: Optional[PRSDataSchema] ccpData: Optional[CCPDataSchema] metadata: ProjectMetadata isConfidential: bool isClaimed: bool currency: str linkfireLinks: Optional[List[LinkfireLinkResponseSchema]] = field(default_factory=list) def __init__( self, project: Project, targets: ProjectTargetSchema, territories: list, shared_status: str, user_roles: List[RoleSchema], metadata: ProjectMetadata, ): self.id = project.id self.name = project.name self.budget = project.budget or 0 self.allocation = project.allocation or 0 self.grasData = GRASDataSchema(project) if project.source == ProjectSource.CCP.value: self.ccpData = CCPDataSchema(project) else: self.prsData = PRSDataSchema(project) self.editDate = project.last_update.created_at self.label = LabelResponseModel(project.label) self.editUser = UserSchema(project.last_update.user) self.createUser = UserSchema(project.created.user) self.currency = project.currency self.targets = targets self.totalSpend = int(project.total_spend) self.status = project.status_id self.territories = [TerritorySchema(territory) for territory in territories] self.userRoles = user_roles self.sharedStatus = shared_status self.notes = project.notes self.startDate = project.start_date self.endDate = project.end_date self.metadata = metadata self.isConfidential = project.is_confidential self.isClaimed = project.is_claimed self.currency = project.currency if project.linkfire_projects: self.linkfireLinks = ProjectLinkfireLinkListMapper.map(project.linkfire_projects) project_response_schema = marshmallow_dataclass.class_schema(ProjectResponseSchema, base_schema=ma.Schema) @dataclass class ProjectCalendarItemResponseSchema: id: int name: str campaignsAmount: int targets: ProjectTargetSchema earliestStartDate: date = field(metadata={"format": "%Y-%m-%d"}) latestEndDate: date = field(metadata={"format": "%Y-%m-%d"}) sharedStatus: Optional[str] def __init__(self, item: ProjectCalendarItemModel, targets: ProjectTargetSchema): self.id = item.id self.name = item.name self.campaignsAmount = item.campaigns_count self.targets = targets self.earliestStartDate = item.earliest_start_date self.latestEndDate = item.latest_end_date self.sharedStatus = item.shared_status @dataclass class ProjectCalendarListSchema: items: List[ProjectCalendarItemResponseSchema] @dataclass class ProjectIdsResponseModel: projectsId: List[int] @dataclass class UserSearchRequestSchema: freeText: Optional[str] = field(metadata={"validate": validate.Length(min=1, max=500)}) @dataclass class UserSearchResponseSchema: id: int email: str name: str labels: List[Optional[LabelSchema]] def __init__(self, user: UsersSearchModel): self.id = user.id self.email = user.email self.name = user.name self.labels = [ LabelSchema(id=label.id, name=label.name, imageUrl=LABEL_IMAGE_SERVICE_URL.format(id=label.id)) for label in user.labels ] @dataclass class ProjectStatusSchema: id: int name: str @dataclass class ProjectHistoryRequestSchema: actionType: Optional[int] query: Optional[str] = field(metadata={"validate": validate.Length(min=1, max=500)}) offset: Optional[int] = field(default=PAGE_OFFSET_DEFAULT, metadata={"validate": validate.Range(min=0)}) limit: Optional[int] = field(default=PAGE_LIMIT_DEFAULT, metadata={"validate": validate.Range(min=1)}) @dataclass class TeamMemberParams: userId: int roleId: int categoryIds: List[int] = field(default_factory=list) @dataclass class TeamMembersUpdateRequestSchema: added: List[TeamMemberParams] = field(default_factory=list) updated: List[TeamMemberParams] = field(default_factory=list) deleted: List[TeamMemberParams] = field(default_factory=list) @dataclass class UpcomingProjectsResponseSchema: id: int name: str targets: ProjectTargetSchema userRoles: List[RoleSchema] budget: float allocation: float startDate: date currency: str isConfidential: bool @dataclass class ProjectArtistSchema: id: str = field(metadata={"validate": validate.Regexp(r"^GRAS_\d*$")}) name: str isPRS: Optional[bool] isUnknown: Optional[bool] type: str = field(metadata={"validate": validate.OneOf(ProjectArtistType.all_values())}, default="Primary") image: Optional[ArtistImageSchema] = None isSony: bool = True def is_primary(self) -> bool: return self.type == "Primary" @dataclass class TrackSchema: id: int isrc: str title: str supplementary: Optional[str] releaseDate: Optional[date] def __init__(self, track: ProductTrack): self.id = track.id self.isrc = track.isrc self.title = track.title self.supplementary = track.title_supplementary self.releaseDate = track.formatted_release_date() @dataclass class ProductFamilySchema: id: int externalId: str title: str releaseDate: Optional[date] tracks: List[TrackSchema] @dataclass class MarketingMixItemSchema: id: int name: str budget: float = 0.0 subtypes: List["MarketingMixItemSchema"] = field(default_factory=list) @dataclass class UnclaimedProjectModelSchema: id: int prsData: Optional[PRSDataSchema] grasData: GRASDataSchema ccpData: Optional[CCPDataSchema] startDate: date endDate: date budget: float allocation: float repertoireOwner: LabelResponseModel artists: Optional[List[ProjectArtistSchema]] currency: str isConfidential: bool isClaimed: bool def __init__(self, project: Project, artists: Optional[List[ProjectArtistSchema]] = None): self.id = project.id if project.source == ProjectSource.CCP.value: self.ccpData = CCPDataSchema(project) else: self.prsData = PRSDataSchema(project) self.grasData = GRASDataSchema(project) self.repertoireOwner = LabelResponseModel(project.label) self.startDate = project.initial_start_date self.endDate = project.end_date self.budget = project.budget or 0 self.allocation = project.allocation or 0 self.currency = project.currency self.artists = artists or [] self.isConfidential = project.is_confidential self.isClaimed = project.is_claimed @dataclass class UnclaimedProjectDatesMetadata: earliest_end_date: Optional[date] = None latest_start_date: Optional[date] = None def __init__(self, metadata): if getattr(metadata, "latest_end_date", None): self.earliest_end_date = metadata.latest_end_date if getattr(metadata, "earliest_start_date", None): self.latest_start_date = metadata.earliest_start_date @dataclass class UnclaimedProjectModelDetailSchema(UnclaimedProjectModelSchema): latestStartDate: Optional[date] earliestEndDate: Optional[date] def __init__( self, project: Project, dates_metadata: UnclaimedProjectDatesMetadata, artists: Optional[List[ProjectArtistSchema]] = None, ): super().__init__(project, artists) self.earliestEndDate = dates_metadata.earliest_end_date self.latestStartDate = dates_metadata.latest_start_date @dataclass class UnclaimedProjectListResponseSchema: metadata: ProjectsCountResponseSchema items: List[UnclaimedProjectModelSchema] = field(default_factory=list) limit: int = field(default_factory=int) offset: int = field(default_factory=int) @dataclass class RecentViewedProjectsQueryParam: limit: int = field(default=MAX_RECENTLY_VIEWED_PROJECTS) label: Optional[int] = field(default=None) recent_viewed_projects_query_param = marshmallow_dataclass.class_schema( RecentViewedProjectsQueryParam, base_schema=ma.Schema ) @dataclass class NewlyAddedProjectsQueryParam: limit: int = field(default=MAX_NEWLY_ADDED_PROJECT, metadata={"validate": validate.Range(min=1)}) label: Optional[int] = field(default=None) newly_added_projects_query_param = marshmallow_dataclass.class_schema( NewlyAddedProjectsQueryParam, base_schema=ma.Schema ) @dataclass class ProjectCardResponseSchema: id: int name: str editDate: Optional[datetime] targets: ProjectTargetSchema isConfidential: bool label: LabelResponseModel def __init__(self, project: Project, targets: ProjectTargetSchema): self.id = project.id if project.is_claimed: self.name = project.name self.editDate = project.last_edit_at else: self.name = project.ccp_title or project.prs_title if project.has_been_claimed: self.editDate = project.last_edit_at self.label = LabelResponseModel(project.label) self.isConfidential = project.is_confidential self.targets = targets @dataclass class RecentViewedProjectsResponseSchema(ProjectCardResponseSchema): def __init__(self, project: Project, targets: ProjectTargetSchema): super().__init__(project, targets) recent_viewed_projects_response_schema = marshmallow_dataclass.class_schema( RecentViewedProjectsResponseSchema, base_schema=ma.Schema ) @dataclass class NewlyAddedProjectsResponseSchema(ProjectCardResponseSchema): def __init__(self, project: Project, targets: ProjectTargetSchema): super().__init__(project, targets) newly_added_projects_response_schema = marshmallow_dataclass.class_schema( NewlyAddedProjectsResponseSchema, base_schema=ma.Schema ) @dataclass class TeamMemberSchema: id: int name: str email: str role: RoleSchema def __init__(self, user): self.id = user.id self.name = user.name self.email = user.email self.role = RoleSchema(UserProjectRoles(user.role)) @dataclass class ProjectSharingResponseSchema: projectTeam: List[TeamMemberSchema] = field(default_factory=list) artistTeamAndAdmins: List[TeamMemberSchema] = field(default_factory=list) approvers: List[TeamMemberSchema] = field(default_factory=list) project_sharing_response_schema = marshmallow_dataclass.class_schema( ProjectSharingResponseSchema, base_schema=ma.Schema ) @dataclass class AddTeamMemberRequestSchema: role_id: int = field(metadata={"data_key": "roleId"}) users_ids: List[int] = field( default_factory=list, metadata={"data_key": "usersIds", "validate": [UniqueValidator(), validate.Length(min=1)]} ) add_team_member_request_schema = marshmallow_dataclass.class_schema( AddTeamMemberRequestSchema, base_schema=ma.Schema ) @dataclass class UpdateDeleteTeamMemberRequestSchema: user_id: int = field(metadata={"data_key": "userId"}) role_id: int = field(metadata={"data_key": "roleId"}) update_delete_team_member_request_schema = marshmallow_dataclass.class_schema( UpdateDeleteTeamMemberRequestSchema, base_schema=ma.Schema )