"""Reports Logic.""" from http import HTTPStatus from typing import Any, Dict, List, Optional import flask from collaborator import config from collaborator.constants import error from collaborator.constants import reports as reports_constants from collaborator.constants.reports import ReportStatus, TriggerType from collaborator.models.ows import ( ows_moneyhub, ows_users, ) from collaborator.models.rds.collaborator_persister import CollaboratorPersister from collaborator.models.rds.report_persister import ReportPersister from collaborator.models.rds.report_run_persister import ReportRunPersister from collaborator.utils import api as api_utils from collaborator.utils import s3 from collaborator.utils.error import OwsError from collaborator.utils.helpers import ( check_collaborators_authorization, check_vendors_authorization, ) from collaborator.utils.sqs import send_bulk_report_trigger_message from collaborator.utils.typing import ( Account, AuthorizedResources, Sort, User, ) def check_can_access_reports_data(authorized_resources, report_ids): """Check if a profile can access report data.""" reports = ReportPersister.get_reports_with_report_ids(report_ids) collaborator_ids = [report["collaborator_id"] for report in reports] collabs_by_id = check_collaborators_authorization( authorized_resources, collaborator_ids ) return reports, collabs_by_id def get_reports( account: Optional[Account] = None, has_transaction: Optional[bool] = None, status: Optional[str] = None, sort_key: Optional[str] = None, sort_direction: Optional[str] = None, limit: int = 0, offset: int = 0, report_run_uuid: Optional[str] = None, collaborator_id: Optional[str] = None, report_id: Optional[str] = None, collaborator_dp_enabled: Optional[bool] = None, ): """Get a list of the reports for an account. Args: account (Account): Account to get reports for has_transaction (bool?): Filter reports who have/haven't transactions status (str?): Filter reports by status sort_key (str?): Key to sort by (defaults to no sorting) sort_direction (str?): Direction to sort by (ASC or DESC) limit (int): Amount of reports to get offset (int): Used to paginate amount of reports collaborator_dp_enabled (bool?): Filter reports by collaborator direct payment enabled status Returns: Response """ sort = Sort(sort_key, sort_direction) if sort_key and sort_direction else None reports = ReportPersister.get_items( account=account, sort=sort, has_transaction=has_transaction, status=status, report_run_uuid=report_run_uuid, collaborator_id=collaborator_id, report_id=report_id, collaborator_dp_enabled=collaborator_dp_enabled, ) # Pagination total_reports = len(reports) if offset != 0: reports = reports[offset:] if limit != 0: reports = reports[:limit] return api_utils.create_paginated_response(reports, total_reports) def get_reports_for_collaborator( collaborator_id: int, account: Account, has_transaction: Optional[bool] = None, status: Optional[str] = None, sort_key: Optional[str] = None, sort_direction: Optional[str] = None, limit: int = 0, offset: int = 0, ): """Retrieve list of report executions corresponding to params. Args: collaborator_id (int): collaborator id account (Account): Account which created the reports. has_transaction (bool?): Filter reports who have/haven't transactions status (str?): Filter reports by status Returns: list: a list of reports """ sort = Sort(sort_key, sort_direction) if sort_key and sort_direction else None reports = ReportPersister.get_items( account, sort=sort, collaborator_id=collaborator_id, has_transaction=has_transaction, status=status, ) # Pagination if offset != 0: reports = reports[offset:] if limit != 0: reports = reports[:limit] return reports def trigger_report_generation( account: Account, user: User, first_period_id: int, last_period_id: int, period_name: str, report_run_name: str, collaborator_ids: Optional[List[int]], ): """Start report generation for the specified periods. Args: first_period_id (int): Start period to create reports for last_period_id (int): End period to create reports for account (Account): Account which is creating the reports client_email (str): The email that the generated reports should be sent. report_run_name (str): Name of the report run. collaborator_ids (list): Optional list of IDs of collaborators to create reports for. period_name(str): Period's name Returns: dict or None """ start_period = min(first_period_id, last_period_id) end_period = max(first_period_id, last_period_id) period_difference = end_period - start_period if period_difference > reports_constants.MAX_REPORT_PERIOD_DIFFERENCE: raise OwsError( code=error.ERROR_CODE_REPORT_PERIOD_RANGE_TOO_LONG, message=error.ERROR_MESSAGE_REPORT_PERIOD_RANGE_TOO_LONG, status=HTTPStatus.BAD_REQUEST, ) identity_metadata = ows_users.get_identity_metadata(user) notification_email = identity_metadata["email"] number_format = identity_metadata["number_format"] currencies_for_period_range = ows_moneyhub.get_currencies_for_period_range( account.id, first_period_id, last_period_id ) if len(currencies_for_period_range) != 1: raise OwsError( code=error.ERROR_CODE_DIFFERENT_CURRENCY_FOR_PERIOD, message=error.ERROR_MESSAGE_DIFFERENT_CURRENCY_FOR_PERIOD, status=HTTPStatus.BAD_REQUEST, ) currency = currencies_for_period_range[0] collab_result = CollaboratorPersister.get_with_split_count( account=account, collaborator_ids=collaborator_ids ) collaborators = [item for item in collab_result if item["splits"] > 0] if len(collaborators) == 0: return None period_ids = ",".join([str(i) for i in range(start_period, end_period + 1)]) correlation_id = flask.g.correlation_id report_run, reports = ReportRunPersister.create_report_run( collaborators=collaborators, period_ids=period_ids, period_name=period_name, name=report_run_name, currency=currency, migrated_to_abacus=True, requestor_identity_uuid=user.id, notification_email=notification_email, number_format=number_format, ) try: send_bulk_report_trigger_message( correlation_id, report_run_uuid=report_run.uuid, ) except Exception: for report in reports: ReportPersister.update_report_status(report.report_id, ReportStatus.ERROR) return {"items": [report.to_dict() for report in reports]} def get_report_download(account: Account, report_id: int): """Get presigned S3 URL for downloading report file. Args: account (Account): Account to which the report belongs report_id(int): RDS report id Returns: dict: Response dict containing presigned URL. Raises: OwsError: 404 if S3 object does not exist. """ report = ReportPersister.get_item(account, report_id) s3_url = report.get("file_location") bucket, key = s3.extract_bucket_path(s3_url) if not s3.object_exists(bucket, key): raise OwsError.not_found( code=error.ERROR_CODE_REPORT_FILE_NOT_FOUND, message=error.ERROR_MESSAGE_REPORT_FILE_NOT_FOUND, ) return {"url": s3.get_presigned_url(bucket, key, config.PRESIGNED_URL_EXPIRES)} def bulk_delete_reports(account: Account, report_ids: list, user: User): """Delete report. Args: account (Account): Account to which the report belongs report_ids (list): list of with the report ids user (User): the user tuple Returns: None """ reports = ReportPersister.get_reports_with_report_ids(report_ids=report_ids) report_run_ids = {report["report_run_id"] for report in reports} report_runs = ReportRunPersister.get_report_runs_by_ids(list(report_run_ids)) has_auto_reports = any( report_run.trigger_type == TriggerType.AUTO for report_run in report_runs ) if has_auto_reports: raise OwsError( code=error.ERROR_CANT_DELETE_AUTOMATIC_REPORT, message=error.ERROR_MESSAGE_CANT_DELETE_AUTOMATIC_REPORT, status=HTTPStatus.BAD_REQUEST, ) ReportPersister.bulk_delete_items(account, report_ids, user) return None def get_reports_with_overlapping_period(report_ids: list): """Get reports with overlapping period. Args: report_ids(list): List of report IDs. Returns: dict: Map of report IDs to array of overlapping report IDs. """ overlaps = ReportPersister.get_report_ids_with_overlapping_period(report_ids) return overlaps def get_latest_reports_by_collaborator_ids( authorized_resources: AuthorizedResources, collaborator_ids: List[int] ): """Get latest reports by collaborator IDs. Args: authorized_resources (AuthorizedResources): Authorized resources collaborator_ids (List[int]): Collaborator IDs Returns: list: Latest reports for the given collaborator IDs. """ results = ReportPersister.get_latest_reports_by_collaborator_ids(collaborator_ids) vendor_ids = list({result.vendor_id for result in results}) authorized_vendor_ids = check_vendors_authorization( authorized_resources, vendor_ids, throw_if_unauthorized=False ) results_by_id = { result.collaborator_id: result for result in results if result.vendor_id in authorized_vendor_ids } return [ { "data": ( {"id": results_by_id[collaborator_id].report_id} if collaborator_id in results_by_id else None ) } for collaborator_id in collaborator_ids ] def get_report_contract_subtotals_dataloader( requested_report_ids: List[int], authorized_resources, ) -> list: """Get non-zero report contract subtotals for given report IDs. Zero-amount entries are excluded. Args: report_ids (List[int]): List of report IDs to get subtotals for contracts. Returns: list """ results = ReportPersister.get_report_contract_subtotals( report_ids=requested_report_ids, ) authorized_vendor_ids = check_vendors_authorization( authorized_resources, [result.vendor_id for result in results], throw_if_unauthorized=False, ) report_contract_subtotals_by_id: Dict[int, List[Dict[str, Any]]] = {} for row in results: if row.vendor_id in authorized_vendor_ids: if row.report_id not in report_contract_subtotals_by_id: report_contract_subtotals_by_id[row.report_id] = [] report_contract_subtotals_by_id[row.report_id].append( { "contract_id": row.contract_id, "subtotal": { "currency": row.currency, "amount": float(row.amount), }, } ) return [ {"data": report_contract_subtotals_by_id.get(report_id, [])} for report_id in requested_report_ids ] def get_report_contract_subtotal_aggregations( report_run_id: int, collaborator_dp_enabled: Optional[bool] = None, ): """Get report contract subtotal aggregations for a given report run ID. Zero-amount entries are excluded. Args: report_run_id (int): Report run ID to get subtotals aggregations for contracts. collaborator_dp_enabled (bool, optional): Filter by collaborator direct payment enabled status. Returns: Row """ return ReportPersister.get_report_contract_subtotal_aggregations( report_run_id, collaborator_dp_enabled )