"""Revenue Analysis Logic.""" from decimal import Decimal from moneyhub import models from moneyhub.constants.constants import ContractType from moneyhub.constants.constants import ONE_DAY from moneyhub.constants.constants import OrderDirection from moneyhub.schemas import PaginatedRevenueByRecordingSchema from moneyhub.schemas.revenue_analysis import AccountRevenueSchema from moneyhub.schemas.revenue_analysis import AccountStatementPeriodRevenueSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByActivityMonthSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByArtistSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByImprintSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByProjectSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByStoreSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueBySubaccountSchema from moneyhub.schemas.revenue_analysis import PaginatedRevenueByTrackSchema from moneyhub.schemas.revenue_analysis import ResponseRevenueByTransactionTypeSchema from moneyhub.schemas.revenue_analysis import RevenueByStatementPeriodSchema from moneyhub.schemas.revenue_analysis import RevenuePayeeCurrencyTotalsSchema from moneyhub.utils.cache import cache from moneyhub.utils.request import create_paginated_response def get_revenue_analysis_by_product( account_id: int, limit: int, offset: int, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, artist_id: int | None, subaccount_id: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, store_ids: list[int] | None, country_codes: list[str] | None, transaction_type_ids: list[int] | None, imprint_ids: list[int] | None, is_subaccount: bool = False, project_id: int | None = None, ) -> dict: """GET Distro Revenue By Product for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) search_term (str): query to search by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by imprint_ids (list[int]): list of imprint IDs to filter by is_subaccount (bool): Whether to include subaccount_revenue project_id (int): Project id to filter by Returns: list: list of distro revenue by product """ subaccount_info = None if not is_subaccount else _get_subaccount_info(subaccount_id) ( records_list, total_records, ) = models.RevenueByProductDistro.get_by_account_id( account_id, limit, offset, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, artist_id, subaccount_id, order_by, order_dir, search_term, store_ids, country_codes, transaction_type_ids, imprint_ids, subaccount_info, project_id ) return create_paginated_response(records_list, total_records) def get_revenue_analysis_by_recording( account_id: int, limit: int, offset: int, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, recording_title: str | None, order_by: str, order_dir: OrderDirection, activity_period_id_start: int | None, activity_period_id_end: int | None, store_ids: list[int] | None, country_codes: list[str] | None, imprint_ids: list[int] | None, transaction_type_ids: list[int] | None, ) -> PaginatedRevenueByRecordingSchema: """GET NR Revenue By Recording for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range recording_title (str): recording name to filter by order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by imprint_ids (list[int]): list of imprint IDs to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by Returns: list: list of nr revenue by recording """ if contract_id: contract_type = models.Contract.get_by_id(contract_id).contract_type else: contract_type = models.Contract.get_contract_type_by_account(account_id).contract_type if contract_type == ContractType.NEIGHBOURING_RIGHTS: recording_list, total_records = models.RevenueByRecordingNr.get_by_account_id( account_id, limit, offset, contract_id, statement_period_id_start, statement_period_id_end, recording_title, order_by, order_dir ) else: recording_list, total_records = models.RevenueByRecordingDistro.get_by_account_id( account_id, limit, offset, contract_id, statement_period_id_start, statement_period_id_end, recording_title, order_by, order_dir, activity_period_id_start, activity_period_id_end, store_ids, country_codes, imprint_ids, transaction_type_ids ) payload = create_paginated_response(recording_list, total_records) return PaginatedRevenueByRecordingSchema(**payload) def get_revenue_analysis_by_statement_period( account_id: int, contract_id: int | None, artist_id: int | None, subaccount_id: int | None, product_id: int | None, track_unique_id: int | None, statement_period_id_start: int, statement_period_id_end: int, is_subaccount: bool = False, activity_period_id_start: int | None = None, activity_period_id_end: int | None = None, store_ids: list[int] | None = None, country_codes: list[str] | None = None, imprint_ids: list[int] | None = None, transaction_type_ids: list[int] | None = None, project_id: int | None = None, ) -> list[RevenueByStatementPeriodSchema]: """Get revenue by statement period for a specified account. Args: account_id (int): ID of the account contract_id (int): ID of a contract to filter by artist_id (int): ID of an artist to filter by subaccount_id (int): the id of a subaccount to filter by product_id (int): the id of a product to filter by track_unique_id (int): the id of a track to filter by statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range is_subaccount (bool): Whether to include subaccount_revenue activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by imprint_ids (list[int]): list of imprint IDs to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by project_id (int | None): project id to filter by Returns: list: list of revenue by statement period """ subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) items = models.RevenueByStatementPeriod.get_by_account_id( account_id, artist_id, subaccount_id, contract_id, product_id, track_unique_id, statement_period_id_start, statement_period_id_end, subaccount_info, activity_period_id_start, activity_period_id_end, store_ids, country_codes, imprint_ids, transaction_type_ids, project_id=project_id, ) # check range and fill in missing periods with 0 values if items and len(items) < (statement_period_id_end - statement_period_id_start + 1): _fill_items(items, statement_period_id_start, statement_period_id_end, is_subaccount) return [RevenueByStatementPeriodSchema.model_validate(item) for item in items] def _fill_items( items: list[RevenueByStatementPeriodSchema], start_period: int, end_period: int, is_subaccount: bool ) -> None: existing_period_ids = {item.statement_period_id for item in items} for period_id in range(start_period, end_period + 1): if period_id not in existing_period_ids: items.append(RevenueByStatementPeriodSchema( statement_period_id=period_id, account_payee_currency=items[0].account_payee_currency, net_revenue_payee_currency=0, gross_revenue_payee_currency=0, subaccount_revenue=0 if is_subaccount else None, net_publishing_revenue_payee_currency=0, gross_publishing_revenue_payee_currency=0 )) def get_revenue_analysis_by_store( account_id: int, artist_id: int | None, subaccount_id: int | None, limit: int, offset: int, contract_id: int | None, product_id: int | None, track_unique_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, is_subaccount: bool = False, activity_period_id_start: int | None = None, activity_period_id_end: int | None = None, country_codes: list[str] | None = None, imprint_ids: list[int] | None = None, transaction_type_ids: list[int] | None = None, store_ids: list[int] | None = None, project_id: int | None = None, ) -> PaginatedRevenueByStoreSchema: """GET Revenue By Store for a specified account ID and optional contract ID. Args: account_id (int): the id of an account artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by contract_id (int): the id of a contract to filter by product_id (int): the id of a product to filter by track_unique_id (int): the id of a track to filter by limit (int): how many records to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): start of the period range statement_period_id_end (int): end of the period range order_by (str): field to sort by (defaults to net revenue payee currency) order_dir (str): direction to sort by (ASC or DESC) search_term (str): query to search by is_subaccount (bool): whether to include subaccount_revenue activity_period_id_start (int): start of the period range activity_period_id_end (int): end of the period range country_codes (list[str] | None): countries to filter by imprint_ids (list[int] | None): ids to filter by transaction_type_ids (list[int] | None): ids to filter by store_ids (list[int] | None): store ids to filter by project_id (int | None): project id to filter by Returns: list: list of revenue by store """ subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) records_list, total_records = models.RevenueByStore.get_by_account_id( account_id, artist_id, subaccount_id, limit, offset, contract_id, product_id, track_unique_id, statement_period_id_start, statement_period_id_end, order_by, order_dir, search_term, subaccount_info, activity_period_id_start, activity_period_id_end, country_codes, imprint_ids, transaction_type_ids, store_ids, project_id=project_id, ) payload = create_paginated_response(records_list, total_records) stores_totals = models.RevenueByStore.get_stores_totals_by_account_id( account_id, artist_id, subaccount_id, contract_id, product_id, track_unique_id, statement_period_id_start, statement_period_id_end, subaccount_info, activity_period_id_start, activity_period_id_end, country_codes, imprint_ids, transaction_type_ids, store_ids, project_id=project_id, ) totals = { 'net_revenue_payee_currency': stores_totals.net_revenue_payee_currency, 'gross_revenue_payee_currency': stores_totals.gross_revenue_payee_currency, } if is_subaccount and hasattr(stores_totals, 'subaccount_revenue'): totals['subaccount_revenue'] = stores_totals.subaccount_revenue payload['totals'] = RevenuePayeeCurrencyTotalsSchema(**totals) return PaginatedRevenueByStoreSchema(**payload) def get_revenue_analysis_by_project( account_id: int, limit: int, offset: int, contract_id: int | None, subaccount_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, is_subaccount: bool = False, activity_period_id_start: int | None = None, activity_period_id_end: int | None = None, store_ids: list[int] | None = None, country_codes: list[str] | None = None, imprint_ids: list[int] | None = None, transaction_type_ids: list[int] | None = None, ) -> PaginatedRevenueByProjectSchema: """Get revenue by project for a specified account. Args: account_id (int): ID of the account limit (int): Maximum records to return offset (int): Number of records to skip contract_id (int | None): Optional contract filter subaccount_id (int | None): Optional subaccount filter statement_period_id_start (int | None): Optional statement period start statement_period_id_end (int | None): Optional statement period end order_by (str): Sort key order_dir (OrderDirection): Sort direction search_term (str | None): Search by project name, code, or id is_subaccount (bool): Whether to include subaccount_revenue activity_period_id_start (int | None): Optional activity period start activity_period_id_end (int | None): Optional activity period end store_ids (list[int] | None): Optional store filter country_codes (list[str] | None): Optional country filter imprint_ids (list[int] | None): Optional imprint filter transaction_type_ids (list[int] | None): Optional transaction type filter Returns: PaginatedRevenueByProjectSchema: paginated project revenue """ subaccount_info = None if not is_subaccount else _get_subaccount_info(subaccount_id) records_list, total_records = models.RevenueByProject.get_by_account_id( account_id, limit, offset, contract_id, subaccount_id, statement_period_id_start, statement_period_id_end, order_by, order_dir, search_term, subaccount_info, activity_period_id_start, activity_period_id_end, country_codes, imprint_ids, transaction_type_ids, store_ids, ) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueByProjectSchema(**payload) def get_revenue_analysis_by_transaction_type( account_id: int, artist_id: int | None, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, subaccount_id: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, is_subaccount: bool = False, product_id: int | None = None, track_unique_id: int | None = None, store_ids: list[int] | None = None, country_codes: list[str] | None = None, imprint_ids: list[int] | None = None, transaction_type_ids: list[int] | None = None, project_id: int | None = None, ) -> ResponseRevenueByTransactionTypeSchema: """Get revenue by transaction type for a given account and optionally contract, period range. Args: account_id (int): ID of the account artist_id (int): the id of an artist to filter by contract_id (int): ID of a contract to filter by statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range subaccount_id (int): Optional ID of a subaccount order_by (str): Key to sort by order_dir (OrderDirection): Direction to sort by search_term (str): query to search by is_subaccount (bool): Whether to include subaccount_revenue product_id (int): The id of the product to filter by track_unique_id (int): the ID of the track to filter by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by imprint_ids (list[int]): list of imprint IDs to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by project_id (int | None): project id to filter by Returns: ResponseRevenueByTransactionTypeSchema: response schema """ subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) items = models.RevenueByTransactionType.get_by_account_id( account_id, artist_id, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, subaccount_id, order_by, order_dir, search_term, product_id, track_unique_id, store_ids, country_codes, imprint_ids, transaction_type_ids, subaccount_info, project_id=project_id, ) transaction_type_totals = ( models.RevenueByTransactionType.get_transaction_types_totals_by_account_id( account_id, artist_id, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, subaccount_id, product_id, track_unique_id, store_ids, country_codes, imprint_ids, transaction_type_ids, subaccount_info, project_id=project_id, ) ) totals = { 'net_revenue_payee_currency': transaction_type_totals.net_revenue_payee_currency, 'gross_revenue_payee_currency': transaction_type_totals.gross_revenue_payee_currency, } if is_subaccount and hasattr(transaction_type_totals, 'subaccount_revenue'): totals['subaccount_revenue'] = transaction_type_totals.subaccount_revenue return ResponseRevenueByTransactionTypeSchema( items=items, totals=RevenuePayeeCurrencyTotalsSchema(**totals) ) def get_revenue_analysis_by_country( account_id: int, artist_id: int | None, subaccount_id: int | None, product_id: int | None, track_unique_id: int | None, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, store_ids: list[int] | None, country_codes: list[str] | None, imprint_ids: list[int] | None, transaction_type_ids: list[int] | None, is_subaccount: bool, project_id: int | None = None, ): """GET Revenue By Country for a specified account ID and optional contract ID. Args: account_id (int): the id of an account artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by product_id (int): the id of a product to filter by track_unique_id (int): the id of a track to filter by contract_id (int): the id of a contract to filter by statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by imprint_ids (list[int]): list of imprint IDs to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by is_subaccount (bool): Whether the request is made by a subaccount project_id (int | None): project id to filter by Returns: list: list of revenue by country """ subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) return models.RevenueByCountry.get_by_account_id( account_id, artist_id, subaccount_id, product_id, track_unique_id, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, store_ids, country_codes, imprint_ids, transaction_type_ids, subaccount_info, project_id=project_id, ) def get_revenue_analysis_by_track( account_id: int, limit: int, offset: int, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, artist_id: int | None, subaccount_id: int | None, product_id: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, store_ids: list[int] | None, country_codes: list[str] | None, transaction_type_ids: list[int] | None, imprint_ids: list[int] | None, is_subaccount: bool, project_id: int | None = None, ) -> PaginatedRevenueByTrackSchema: """GET Revenue By Track for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by product_id (int): the id of a product to filter by order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) search_term (str): query to search by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by imprint_ids (list[int]): list of imprint IDs to filter by is_subaccount (bool): Whether the request is made by a subaccount Returns: PaginatedRevenueByTrackSchema: list of revenue by track """ subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) ( records_list, total_records, ) = models.RevenueByTrack.get_by_account_id( account_id, limit, offset, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, artist_id, subaccount_id, product_id, order_by, order_dir, search_term, store_ids, country_codes, transaction_type_ids, imprint_ids, subaccount_info, project_id ) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueByTrackSchema(**payload) def get_revenue_analysis_by_artist( account_id: int, limit: int, offset: int, contract_id: int | None, subaccount_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, store_ids: list[int] | None, country_codes: list[str] | None, transaction_type_ids: list[int] | None, imprint_ids: list[int] | None, is_subaccount: bool = False ) -> PaginatedRevenueByArtistSchema: """GET Revenue by artist for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by subaccount_id (int): the id of a subaccount to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) search_term (str): query to search by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by imprint_ids (list[int]): list of imprint IDs to filter by is_subaccount (bool): flag of check if query is from a subaccount Returns: PaginatedRevenueByArtistSchema: list of revenue by artist """ subaccount_info = None if not is_subaccount else _get_subaccount_info(subaccount_id) ( records_list, total_records, ) = models.RevenueByArtist.get_by_account_id( account_id, limit, offset, contract_id, subaccount_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, order_by, order_dir, search_term, store_ids, country_codes, transaction_type_ids, imprint_ids, subaccount_info ) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueByArtistSchema(**payload) def get_revenue_analysis_by_subaccount( account_id: int, limit: int, offset: int, artist_id: int | None, contract_id: int | None, store_ids: list[int] | None, activity_period_id_start: int | None, activity_period_id_end: int | None, imprint_ids: list[int] | None, transaction_type_ids: list[int] | None, country_codes: list[str] | None, statement_period_id_start: int | None, statement_period_id_end: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, ) -> PaginatedRevenueBySubaccountSchema: """GET Revenue by subaccount for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). artist_id (int): the id of an artist to filter by store_ids (list): Optional stores to filter by activity_period_id_start (int): Optional activity period range start activity_period_id_end (int): Optional activity period range end imprint_ids (list): Optional imprints to filter by transaction_type_ids (list): Optional transaction types to filter by country_codes (list): Optional country codes to filter by statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) search_term (str): query to search by Returns: PaginatedRevenueBySubaccountSchema: list of revenue by subaccount """ ( records_list, total_records, ) = models.RevenueBySubaccount.get_by_account_id( account_id, limit, offset, artist_id, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, store_ids, imprint_ids, transaction_type_ids, country_codes, order_by, order_dir, search_term) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueBySubaccountSchema(**payload) def get_revenue_analysis_by_imprint( account_id: int, artist_id: int | None, limit: int, offset: int, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, subaccount_id: int | None, order_by: str, order_dir: OrderDirection, search_term: str | None, store_ids: list[int] | None, country_codes: list[str] | None, imprint_ids: list[int] | None, transaction_type_ids: list[int] | None, is_subaccount: bool = False, ) -> PaginatedRevenueByImprintSchema: """GET Revenue by imprint for a specified account ID and optional contract ID. Args: account_id (int): the id of an account artist_id (int): the id of an artist to filter by contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range subaccount_id (int): the id of a subaccount order_by (str): Key to sort by (defaults to no sorting) order_dir (str): Direction to sort by (ASC or DESC) search_term (str | int): query to search by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by is_subaccount (bool): Whether to include subaccount_revenue Returns: PaginatedRevenueByImprintSchema: list of revenue by imprint """ subaccount_info = None if not is_subaccount else _get_subaccount_info(subaccount_id) ( records_list, total_records, ) = models.RevenueByImprint.get_by_account_id( account_id, artist_id, limit, offset, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, subaccount_id, order_by, order_dir, search_term, store_ids, country_codes, imprint_ids, transaction_type_ids, subaccount_info ) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueByImprintSchema(**payload) def get_revenue_analysis_by_activity_month( account_id: int, limit: int | None, offset: int | None, artist_id: int | None, subaccount_id: int | None, product_id: int | None, track_unique_id: int | None, contract_id: int | None, statement_period_id_start: int | None, statement_period_id_end: int | None, activity_period_id_start: int | None, activity_period_id_end: int | None, imprint_ids: list[int] | None, store_ids: list[int] | None, country_codes: list[str] | None, transaction_type_ids: list[int] | None, is_subaccount: bool = False, project_id: int | None = None, ) -> PaginatedRevenueByActivityMonthSchema: """GET Revenue by activity month for a specified account ID and optional contract ID. Args: account_id (int): the id of an account contract_id (int): the id of a contract to filter by limit (int): how many entities to retrieve offset (int): the offset (for pagination) artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by product_id (int): the id of a product to filter by track_unique_id (int): the id of a track to filter by statement_period_id_start (int): Start of the period range statement_period_id_end (int): End of the period range activity_period_id_start (int): Start of the activity period range activity_period_id_end (int): End of the activity period range imprint_ids (list[int]): list of imprint IDs to filter by store_ids (list[int]): list of store IDs to filter by country_codes (list[str]): list of country codes to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by is_subaccount (bool): bool flag to check who is making the request project_id (int | None): project id to filter by Returns: PaginatedRevenueByActivityMonthSchema: list of revenue by activity month """ subaccount_info = None if not is_subaccount else _get_subaccount_info(subaccount_id) ( records_list, total_records, ) = models.RevenueByActivityMonth.get_revenue_by_activity_month_by_account_id( account_id, limit, offset, artist_id, subaccount_id, product_id, track_unique_id, contract_id, statement_period_id_start, statement_period_id_end, activity_period_id_start, activity_period_id_end, imprint_ids, store_ids, country_codes, transaction_type_ids, subaccount_info, project_id=project_id, ) payload = create_paginated_response(records_list, total_records) return PaginatedRevenueByActivityMonthSchema(**payload) def get_revenue_by_account( account_id: int, artist_id: int | None, subaccount_id: int | None, is_subaccount: bool = False, ) -> AccountRevenueSchema: """Get revenue info for a specified account. Args: account_id (int): The id of an account artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by is_subaccount (bool): Whether to include subaccount_revenue Returns: dict: Account revenue details """ visible_periods = models.AccountStatementPeriods.get_activity_statement_period_ids( account_id=account_id) last_statement_period_id = visible_periods[-1] subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) currency_changes = models.RevenueByStatementPeriod.get_currencies_periods( account_id, visible_periods, subaccount_id, artist_id) if not currency_changes: # account has no revenue account_payment_term = models.AccountPaymentTerm.get_by_account_id(account_id) no_revenue_object = { 'since_statement_period_id': None, 'currency_code': account_payment_term.currency_code, 'gross_revenue_payee_currency': 0, 'net_revenue_payee_currency': 0 } return AccountRevenueSchema(**no_revenue_object) latest_currency = currency_changes[0].account_payee_currency since_statement_period_id = int(currency_changes[-1].statement_period_id) # get period where change occurred for row in currency_changes: currency, statement_period_id = row.account_payee_currency, row.statement_period_id if currency != latest_currency: break since_statement_period_id = int(statement_period_id) revenue = models.RevenueByStatementPeriod.get_revenue_total_for_account_statement_periods( account_id, since_statement_period_id, last_statement_period_id, artist_id, subaccount_id, subaccount_info ) revenue_schema = AccountRevenueSchema(**revenue) revenue_schema.since_statement_period_id = since_statement_period_id return revenue_schema def get_latest_revenue_by_account_id( account_id: int, statement_period: int, artist_id: int | None, subaccount_id: int | None, product_id: int | None, track_unique_id: int | None, is_subaccount: bool = False, project_id: int | None = None, ) -> AccountStatementPeriodRevenueSchema: """Get revenue latest info for a specified account. Args: account_id (int): The id of an account statement_period (int): the id of the statement period artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by product_id (int): the id of a product to filter by track_unique_id (int): the id of a track to filter by is_subaccount (bool): Whether to include subaccount_revenue project_id (int | None): project id to filter by Returns: dict: Account revenue details """ payment_term = models.AccountPaymentTerm.get_by_account_id(account_id) if not payment_term: return None latest_statement_period = models.StatementPeriod.get_by_id(statement_period) subaccount_info = None if is_subaccount: subaccount_info = _get_subaccount_info(subaccount_id) revenue = models.RevenueByStatementPeriod.get_by_account_id( account_id=account_id, contract_id=None, artist_id=artist_id, subaccount_id=subaccount_id, product_id=product_id, track_unique_id=track_unique_id, statement_period_id_start=latest_statement_period.statement_period_id, statement_period_id_end=latest_statement_period.statement_period_id, subaccount_info=subaccount_info, project_id=project_id ) result = { 'currency_code': None, 'statement_period_id': latest_statement_period.statement_period_id, 'gross_revenue_payee_currency': Decimal(0), 'net_revenue_payee_currency': Decimal(0), 'gross_publishing_revenue_payee_currency': None, 'net_publishing_revenue_payee_currency': None, 'subaccount_revenue': None } if (revenue): result['currency_code'] = revenue[0].account_payee_currency result['gross_revenue_payee_currency'] = revenue[0].gross_revenue_payee_currency result['net_revenue_payee_currency'] = revenue[0].net_revenue_payee_currency result['gross_publishing_revenue_payee_currency'] = \ revenue[0].gross_publishing_revenue_payee_currency result['net_publishing_revenue_payee_currency'] = \ revenue[0].net_publishing_revenue_payee_currency if (is_subaccount): result['subaccount_revenue'] = revenue[0].subaccount_revenue if not result['currency_code']: result['currency_code'] = payment_term.currency_code return AccountStatementPeriodRevenueSchema(**result) @cache(ttl=ONE_DAY) def _get_subaccount_info(subaccount_id: int) -> object: """Get a subaccount's commission override and split type. Args: subaccount_id(int): ID of subaccount Returns: DimSubaccount row object """ return models.DimSubaccount.get_by_id(subaccount_id)