"""RevenueByActivityMonth Model.""" from sqlalchemy import Column from sqlalchemy import desc from sqlalchemy import func from sqlalchemy import Numeric from sqlalchemy import String from sqlalchemy.engine import Row from moneyhub.models.revenue_base import apply_filters from moneyhub.models.revenue_base import apply_subaccount_revenue from moneyhub.models.snowflake_base import BaseModel class RevenueByActivityMonth(BaseModel): """Revenue By Activity Month model.""" __tablename__ = 'revenue_by_activity_month_dbt' account_id = Column(Numeric(12, 0), nullable=False, primary_key=True) account_payee_currency = Column(String(50), nullable=False) activity_period_id = Column(Numeric(32, 0), nullable=False, primary_key=True) activity_month_name = Column(String(50), nullable=False) artist_id = Column(Numeric(32, 0), nullable=False, primary_key=True) contract_id = Column(Numeric(12, 0), nullable=False, primary_key=True) 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) imprint_id = Column(Numeric(32, 0), nullable=True) store_id = Column(Numeric(32, 0), nullable=True) project_id = Column(Numeric(32, 0), nullable=True) country_code = Column(String(3), nullable=True) transaction_type_id = Column(Numeric(32, 0), nullable=True) statement_period_id = Column(Numeric(38, 0), nullable=False, primary_key=True) net_share_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_activity_periods_by_account_id( cls, account_id: int, subaccount_id: int | None, activity_period_ids: list[int] | None, ) -> list: """GET a list of all activity months by account ID. Args: account_id (int): Account to get activity periods for. subaccount_id (int): Optional subaccount to filter by. activity_period_ids (list): Optional list of activity period IDs to filter by. Returns: list: List of activity periods. """ filters = [(cls.account_id == account_id)] if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if activity_period_ids: filters.append(cls.activity_period_id.in_(activity_period_ids)) return cls.query\ .with_entities( cls.activity_period_id, cls.activity_month_name, )\ .distinct()\ .filter(*filters)\ .order_by(cls.activity_period_id)\ .all() @classmethod def get_revenue_by_activity_month_by_account_id( cls, account_id: int, limit: int, offset: 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, imprint_ids: list[int] | None = None, store_ids: list[int] | None = None, country_codes: list[str] | None = None, transaction_type_ids: list[int] | None = None, subaccount_info: Row | None = None, project_id: int | None = None, ) -> tuple[list, int]: """GET list of revenue by activity month for a specified account ID and contract ID. Args: account_id (int): the id of an account limit (int): how many entities to retrieve offset (int): the offset (for pagination) artist_id (int | None): the id of an artist to filter by subaccount_id (int | None): the id of a subaccount to filter by product_id (int | None): the id of a product to filter by track_unique_id (int | None): the id of a track to filter by contract_id (int | None): the id of a contract to filter by statement_period_id_start (int | None): Start of the period range statement_period_id_end (int | None): End of the period range activity_period_id_start (int | None): Start of the activity period range activity_period_id_end (int | None): End of the activity period range imprint_ids (list[int] | None): list of imprint IDs to filter by store_ids (list[int] | None): list of store IDs to filter by country_codes (list[str] | None): list of country codes to filter by transaction_type_ids (list[int] | None): list of transaction type IDs to filter by subaccount_info (Row | None): subaccount info for revenue calculation project_id (int | None): project id to filter by Returns: Tuple[list, int]: list of revenue by activity month and total record count """ filters = [(cls.account_id == account_id)] group_by = [ cls.activity_period_id, cls.activity_month_name, cls.account_payee_currency, ] with_entities = [ cls.activity_period_id, cls.activity_month_name, cls.account_payee_currency, func.sum(cls.net_share_payee_currency).label('net_share_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'), func.count().over().label('total_records') ] apply_subaccount_revenue(subaccount_info, with_entities, cls) if artist_id: filters.append(cls.artist_id == artist_id) if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if contract_id: filters.append(cls.contract_id == contract_id) apply_filters( cls=cls, filters=filters, country_codes=country_codes, imprint_ids=imprint_ids, product_id=product_id, project_id=project_id, statement_period_id_start=statement_period_id_start, statement_period_id_end=statement_period_id_end, activity_period_id_start=activity_period_id_start, activity_period_id_end=activity_period_id_end, store_ids=store_ids, track_unique_id=track_unique_id, transaction_type_ids=transaction_type_ids, ) query = cls.query.with_entities(*with_entities).filter(*filters).group_by( *group_by).order_by(desc(cls.activity_period_id)) records = query.limit(limit).offset(offset).all() if limit else query.all() total_records = int(records[0].total_records) if records else 0 return records, total_records