from datetime import date from typing import List, Iterator, Optional from services.exporter import ExportItem, CSVExporter from constants.project_statuses import PROJECT_STATUSES_MAPPING from projects.services.projects_service import ProjectsService from projects.schemas import ProjectListQueryParams from models.artists import Artist from models.projects import ProjectEntityType class ProjectsExporter: def __init__(self, service: ProjectsService = ProjectsService()): self.service = service def export_projects_csv( self, user_id: int, filters: ProjectListQueryParams, artist_external_id: Optional[str] = None ) -> str: """Export list of projects to CSV. Args: user_id (id): Resulting list will contain only projects available to this user. A project is available to a user if the user is a member of project's label or if project is shared with this user. filters (dict): Dictionary with optional filters. Accepts the same set of filters as get_project_list() Returns: str: Content of a CSV file as string. """ # This implementation fetches all projects in a single query. It is the simplest and optimal solution while # there aren't many projects. When their number grows more than ten thousands a testing on real data # can be done to determine what kind of optimization is needed if any. exporter = CSVExporter() projects = self.service.get_project_export_list(user_id, filters, artist_external_id) for c in projects: export_item = ExportItem() start_date = date.strftime(c.earliest_start_date, "%Y-%m-%d") if c.earliest_start_date else "" end_date = date.strftime(c.latest_end_date, "%Y-%m-%d") if c.latest_end_date else "" export_item.add_item("Project Name", 100, c.name) export_item.add_item("Project Status", 200, PROJECT_STATUSES_MAPPING[c.status]) export_item.add_item("Project Start Date", 300, start_date) export_item.add_item("Project End Date", 400, end_date) export_item.add_item("Project Budget", 500, c.planned_budget) export_item.add_item("Project Budget Spent", 600, c.total_spend) if isinstance(c.targets[0].entity, Artist): for i, item in enumerate(self.filter_by_type(c.targets, ProjectEntityType.PRIMARY_ARTIST)): export_item.add_item(f"Primary Artist {i+1}", 700 + i, item.entity.name) names = ", ".join( [item.entity.name for item in self.filter_by_type(c.targets, ProjectEntityType.FEATURED_ARTIST)] ) if names: export_item.add_item("Featured Artists", 750, names) else: names = ", ".join([item.entity.name for item in c.targets]) export_item.add_item("Playlists", 700, names) exporter.add_export_item(export_item) return exporter.export() def filter_by_type(self, targets: List[Artist], entity_type: ProjectEntityType) -> Iterator[Artist]: return filter(lambda x: entity_type == ProjectEntityType(x.entity_type), targets)