from dataclasses import dataclass from typing import List, Optional from projects.constants import ProjectTargetTypes from models import Polymorphable from models.projects import Project, ProjectTargetItem, ProjectEntityType from models.user import User from services.users_service import UsersService from services.project_history.history_service import HistoryService from services.project_history.schemas import HistoryTerritory from shared.schemas import ProjectLinkfireLinkListMapper, LinkfireLinkResponseSchema from projects.repositories.projects_repository import ProjectsRepository @dataclass class ProjectTargetHistoryItem: entity: Polymorphable entity_type: int is_deleted: bool def __init__(self, item: ProjectTargetItem): self.entity = item.entity self.entity_type = item.entity_type self.is_deleted = item.is_deleted def is_primary_artist(self) -> bool: return self.entity_type == ProjectEntityType.PRIMARY_ARTIST.value def is_featured_artist(self) -> bool: return self.entity_type == ProjectEntityType.FEATURED_ARTIST.value def is_playlist(self) -> bool: return self.entity_type == ProjectEntityType.PLAYLIST.value @dataclass class ProjectHistoryState: name: str targets_type: str target_items: List[ProjectTargetHistoryItem] budget: float notes: Optional[str] linkfire_links: List[LinkfireLinkResponseSchema] def not_deleted_items(self): return list(filter(lambda x: not x.is_deleted, self.target_items)) @dataclass class CampaignTerritoriesHistoryState: territories: List[HistoryTerritory] def __init__(self, territories: List[HistoryTerritory]): self.territories = territories class ProjectHistoryService: users_service = UsersService() history_service = HistoryService() projects_repository = ProjectsRepository() def get_project_state(self, project: Project) -> ProjectHistoryState: return ProjectHistoryState( name=project.name, targets_type=project.target_items[0].entity.type if project.target_items else None, target_items=[ProjectTargetHistoryItem(item) for item in project.target_items], budget=project.budget, notes=project.notes, linkfire_links=ProjectLinkfireLinkListMapper.map(project.linkfire_projects), ) def log_project_changes(self, project: Project, project_state: ProjectHistoryState, user_id: Optional[int]): type_string = project.target_items[0].entity.type if project.target_items else None old_type = ProjectTargetTypes.from_value(project_state.targets_type) new_type = ProjectTargetTypes.from_value(type_string) old_playlists = project_state.target_items if old_type == ProjectTargetTypes.PLAYLIST else [] new_playlists = project.target_items if new_type == ProjectTargetTypes.PLAYLIST else [] old_artists = project_state.not_deleted_items() if old_type == ProjectTargetTypes.ARTIST else [] new_artists = project.not_deleted_items() if new_type == ProjectTargetTypes.ARTIST else [] self.__log_project_name_changed_if_needed(project, project_state.name, project.name, user_id) self.__log_project_budget_changed_if_needed(project, project_state.budget, project.budget, user_id) self.__log_project_linkfire_links_changed_if_needed( project, project_state.linkfire_links, ProjectLinkfireLinkListMapper.map(project.linkfire_projects), user_id ) if old_type and new_type: self.__log_project_target_type_changed_if_needed(project, project_state.targets_type, type_string, user_id) self.__log_project_primary_artists_changed_if_needed(project, old_artists, new_artists, user_id) self.__log_project_featured_artists_changed_if_needed(project, old_artists, new_artists, user_id) self.__log_project_playlists_changed_if_needed(project, old_playlists, new_playlists, user_id) self.__log_project_notes_changes_if_needed(project, project_state.notes, project.notes, user_id) def log_project_created(self, project: Project, user_id: Optional[int]): self.history_service.log_project_created(project, user_id) if project.target_items: targets_type = ProjectTargetTypes.from_value(project.target_items[0].entity.type) if targets_type == ProjectTargetTypes.ARTIST: self.__log_project_primary_artists_changed_if_needed(project, [], project.target_items, user_id) self.__log_project_featured_artists_changed_if_needed(project, [], project.target_items, user_id) else: self.__log_project_playlists_changed_if_needed(project, [], project.target_items, user_id) if project.notes is not None and len(project.notes) > 0: self.history_service.log_project_notes_updated(project, None, project.notes, user_id) def __log_project_budget_changed_if_needed( self, project: Project, old_budget: float, new_budget: float, user_id: Optional[int] ): if new_budget != old_budget: self.history_service.log_project_budget_updated(project, old_budget, new_budget, user_id) def __log_project_notes_changes_if_needed( self, project: Project, old_notes: Optional[str], notes: Optional[str], user_id: Optional[int] ): if old_notes != notes: self.history_service.log_project_notes_updated(project, old_notes, notes, user_id) def __log_project_name_changed_if_needed( self, project: Project, old_name: str, new_name: str, user_id: Optional[int] ): if old_name != new_name: self.history_service.log_project_name_updated(project.id, old_name, new_name, user_id) def __log_project_target_type_changed_if_needed( self, project: Project, old_type: str, new_type: str, user_id: Optional[int] ): if old_type != new_type: self.history_service.log_project_target_type_updated(project.id, old_type, new_type, user_id) def __log_project_primary_artists_changed_if_needed( self, project: Project, old_artists: List[ProjectTargetItem], new_artists: List[ProjectTargetItem], user_id: Optional[int], ): old_primary_artists = list(map(lambda x: x.entity, filter(lambda x: x.is_primary_artist(), old_artists))) new_primary_artists = list(map(lambda x: x.entity, filter(lambda x: x.is_primary_artist(), new_artists))) added_artists = sorted(list(set(new_primary_artists) - set(old_primary_artists)), key=lambda x: x.name) removed_artists = sorted(list(set(old_primary_artists) - set(new_primary_artists)), key=lambda x: x.name) if added_artists: self.history_service.log_added_primary_artists(project.id, added_artists, user_id) if removed_artists: self.history_service.log_removed_primary_artists(project.id, removed_artists, user_id) def __log_project_featured_artists_changed_if_needed( self, project: Project, old_artists: List[ProjectTargetItem], new_artists: List[ProjectTargetItem], user_id: Optional[int], ): old_featured_artists = list(map(lambda x: x.entity, filter(lambda x: x.is_featured_artist(), old_artists))) new_featured_artists = list(map(lambda x: x.entity, filter(lambda x: x.is_featured_artist(), new_artists))) added_artists = sorted(list(set(new_featured_artists) - set(old_featured_artists)), key=lambda x: x.name) removed_artists = sorted(list(set(old_featured_artists) - set(new_featured_artists)), key=lambda x: x.name) if added_artists: self.history_service.log_added_featured_artists(project.id, added_artists, user_id) if removed_artists: self.history_service.log_removed_featured_artists(project.id, removed_artists, user_id) def __log_project_playlists_changed_if_needed( self, project: Project, old_items: List[ProjectTargetItem], new_items: List[ProjectTargetItem], user_id: Optional[int], ): old_playlists = list(map(lambda x: x.entity, old_items)) new_playlists = list(map(lambda x: x.entity, new_items)) added = set(new_playlists) - set(old_playlists) removed = set(old_playlists) - set(new_playlists) if added: self.history_service.log_added_playlists( project.id, sorted(added, key=lambda playlist: playlist.name), user_id ) if removed: self.history_service.log_removed_playlists( project.id, sorted(removed, key=lambda playlist: playlist.name), user_id ) def __log_project_linkfire_links_changed_if_needed( self, project: Project, old_value: List, new_value: List, user_id: Optional[int] ): if old_value != new_value: self.history_service.log_project_linkfire_changes(project, old_value, new_value, user_id) def log_project_claimed(self, project: Project, project_state: ProjectHistoryState, user_id: Optional[int]): self.history_service.log_claimed_project(project, user_id) default_type = "Artist" old_type = ProjectTargetTypes.from_value(project_state.targets_type) type_string = project.target_items[0].entity.type new_type = ProjectTargetTypes.from_value(type_string) new_playlists = project.target_items if new_type == ProjectTargetTypes.PLAYLIST else [] old_artists = project_state.not_deleted_items() if old_type == ProjectTargetTypes.ARTIST else [] new_artists = project.not_deleted_items() if new_type == ProjectTargetTypes.ARTIST else [] if project.notes: self.history_service.log_project_notes_updated(project, None, project.notes, user_id) self.__log_project_target_type_changed_if_needed(project, default_type, type_string, user_id) self.__log_project_primary_artists_changed_if_needed(project, old_artists, new_artists, user_id) self.__log_project_featured_artists_changed_if_needed(project, old_artists, new_artists, user_id) self.__log_project_playlists_changed_if_needed(project, [], new_playlists, user_id)