from typing import List from flask import g from models.projects import Project from services.recent_search.recent_search_repository import RecentSearchesRepository from services.recent_search.schemas import RecentSearchItemResponse, ProjectSearchModel from models.recent_search_item import RecentSearchItem, RecentSearchItemType from shared.schemas import ArtistSchema from projects.services.project_target_items_maper import ProjectTargetItemsMapper from services.recent_search.campaign_mapper_service import CampaignMapperService from services.permission_role_service import PermissionsByRoleService class RecentSearchesService: repository = RecentSearchesRepository() target_items_mapper = ProjectTargetItemsMapper() campaign_mapper = CampaignMapperService() permissions_role_service = PermissionsByRoleService() def get_recent_searches(self, user_id: int, params) -> List[RecentSearchItemResponse]: items = self.repository.get_recent_searches_for_user(user_id, params.limit) return [ RecentSearchItemResponse(id=item.id, type=item.type, entity=self.__map_recent_search_item(item)) for item in items ] def __map_recent_search_item(self, item: RecentSearchItem): if item.type == RecentSearchItemType.ARTIST.value: return ArtistSchema.from_db_model(item.artist) if item.type == RecentSearchItemType.PROJECT.value: return self.__map_project(item.project) if item.type == RecentSearchItemType.CAMPAIGN.value: return self.campaign_mapper.map_entity(item.campaign) def __map_project(self, project: Project): target_items = self.target_items_mapper.map_project_target_items_from_model(project.not_deleted_items()) is_claimed = project.is_claimed roles = self.permissions_role_service.get_roles_for_project(user_id=g.user_id, project=project) return ProjectSearchModel( id=project.id, name=project.display_name, startDate=project.start_date, targets=target_items, isClaimed=is_claimed, isConfidential=project.is_confidential, userRoles=roles, editDate=project.last_edit_at, ) def track_opened_project_details(self, project_id: int, user_id: int, params: dict): self.repository.track_recently_viewed_project(project_id, user_id) self.repository.clean_old_recently_viewed_projects(user_id) if self.__should_track(params): self.repository.track_opened_project_details(project_id, user_id) self.repository.clean_old_searches(user_id) def track_opened_campaign_details(self, campaign_uuid: str, user_id: int, params: dict): if self.__should_track(params): self.repository.track_opened_campaign_details(campaign_uuid, user_id) self.repository.clean_old_searches(user_id) def track_opened_artist_details(self, artist_external_id: str, user_id: int, params: dict): if self.__should_track(params): self.repository.track_opened_artist_details(artist_external_id, user_id) self.repository.clean_old_searches(user_id) def __should_track(self, params) -> bool: return params.get("fromSearch") is not None