"""RevenueByStatementPeriod Model.""" from sqlalchemy import Column from sqlalchemy import desc from sqlalchemy import Enum from sqlalchemy import func from sqlalchemy import Numeric from sqlalchemy import String from sqlalchemy.engine.row import Row from moneyhub.constants.constants import StatementPeriodStatus from moneyhub.constants.features import FEATURE_PUBLISHING_PHASE_ONE from moneyhub.models.revenue_base import apply_filters from moneyhub.models.revenue_base import apply_subaccount_revenue from moneyhub.models.snowflake_base import BaseModel from moneyhub.utils.features import is_feature_enabled class RevenueByStatementPeriod(BaseModel): """Revenue By Statement Period model.""" __tablename__ = 'revenue_by_statement_period_dbt' account_id = Column(Numeric(12, 0), nullable=False, primary_key=True) artist_id = Column(Numeric(32, 0), nullable=True, primary_key=True) contract_id = Column(Numeric(12, 0), nullable=True, primary_key=True) statement_period_id = Column(Numeric(38, 0), nullable=False, primary_key=True) statement_period_name = Column(String(180), nullable=False) statement_period_status = Column( Enum( *StatementPeriodStatus, name='statement_period_status', create_type=False ), nullable=False ) subaccount_id = Column(Numeric(32, 0), nullable=True, primary_key=True) product_id = Column(Numeric(32, 0), nullable=True) track_unique_id = Column(Numeric(32, 0), nullable=True) activity_period_id = Column(Numeric(38, 0), nullable=True) country_code = Column(String(255), nullable=True) imprint_id = Column(Numeric(32, 0), nullable=True) store_id = Column(Numeric(32, 0), nullable=True) project_id = Column(Numeric(32, 0), nullable=True) transaction_type_id = Column(Numeric(32, 0), nullable=True) account_payee_currency = Column(String(50), nullable=False) net_revenue_payee_currency = Column(Numeric(36, 12), nullable=True) gross_revenue_payee_currency = Column(Numeric(36, 12), nullable=True) net_publishing_revenue_payee_currency = Column(Numeric(36, 12), nullable=True) gross_publishing_revenue_payee_currency = Column(Numeric(36, 12), nullable=True) @classmethod def get_by_account_id( cls, account_id: int, artist_id: int | None, subaccount_id: int | None, 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, subaccount_info: dict | None, 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: """GET list of revenue by statement period for a specified account 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 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 subaccount_info (dict): Subaccount information 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 """ filters = [(cls.account_id == account_id)] group_by = [ cls.account_payee_currency, cls.statement_period_id, ] with_entities = [ cls.account_payee_currency, cls.statement_period_id, func.sum(cls.net_revenue_payee_currency).label('net_revenue_payee_currency'), func.sum(cls.gross_revenue_payee_currency).label('gross_revenue_payee_currency'), func.sum(cls.net_publishing_revenue_payee_currency).label( 'net_publishing_revenue_payee_currency'), func.sum(cls.gross_publishing_revenue_payee_currency).label( 'gross_publishing_revenue_payee_currency'), ] apply_subaccount_revenue(subaccount_info, with_entities, cls) apply_filters( cls, filters, artist_id=artist_id, statement_period_id_start=statement_period_id_start, statement_period_id_end=statement_period_id_end, track_unique_id=track_unique_id, subaccount_id=subaccount_id, contract_id=contract_id, product_id=product_id, project_id=project_id, activity_period_id_start=activity_period_id_start, activity_period_id_end=activity_period_id_end, store_ids=store_ids, country_codes=country_codes, imprint_ids=imprint_ids, transaction_type_ids=transaction_type_ids, ) return cls.query \ .with_entities(*with_entities) \ .filter(*filters) \ .group_by(*group_by) \ .all() @classmethod def get_currencies_periods( cls, account_id: int, statement_period_ids: list, subaccount_id: int | None, artist_id: int | None, ) -> Row | None: """Get currencies and by statement periods for an account. Args: account_id (int): Account to get balance for. statement_period_ids (list): List of statement periods to filter by subaccount_id (int): the id of a subaccount to filter by artist_id (int): the id of an artist to filter by Returns: Row: SqlAlchemy row containing currency information. """ filters = [cls.account_id == account_id, cls.statement_period_id.in_(statement_period_ids)] if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if artist_id: filters.append(cls.artist_id == artist_id) with_entities = [ cls.account_payee_currency, cls.statement_period_id, ] return ( cls.query .with_entities(*with_entities) .filter(*filters) .order_by(desc(cls.statement_period_id)) .distinct() .all() ) @classmethod def get_revenue_total_for_account_statement_periods( cls, account_id: int, statement_period_id_start: int, statement_period_id_end: int, artist_id: int | None, subaccount_id: int | None, subaccount_info: Row | None, ) -> Row | None: """Get the aggregate revenue for a given account and list of statement period ids. Args: account_id (int): Account to get balance for. statement_period_id_start (int): start of statement period to filter by statement_period_id_end (int): end of statement period to filter by artist_id (int): the id of an artist to filter by subaccount_id (int): the id of a subaccount to filter by subaccount_info (dict): Subaccount information Returns: Row: SqlAlchemy row containing revenue information. """ filters = [ cls.account_id == account_id, cls.statement_period_id.between(statement_period_id_start, statement_period_id_end) ] if artist_id: filters.append(cls.artist_id == artist_id) if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) with_entities = [ cls.account_payee_currency.label('currency_code'), ] if is_feature_enabled(FEATURE_PUBLISHING_PHASE_ONE): with_entities += [ ( func.coalesce(func.sum(cls.gross_revenue_payee_currency), 0) + func.coalesce(func.sum(cls.gross_publishing_revenue_payee_currency), 0) ).label('gross_revenue_payee_currency'), ( func.coalesce(func.sum(cls.net_revenue_payee_currency), 0) + func.coalesce(func.sum(cls.net_publishing_revenue_payee_currency), 0) ).label('net_revenue_payee_currency') ] else: with_entities += [ func.sum(cls.gross_revenue_payee_currency).label('gross_revenue_payee_currency'), func.sum(cls.net_revenue_payee_currency).label('net_revenue_payee_currency') ] apply_subaccount_revenue(subaccount_info, with_entities, cls) return cls.query.with_entities( *with_entities ).filter(*filters).group_by(cls.account_payee_currency).first()