"""Revenue by transaction type model.""" from sqlalchemy import asc from sqlalchemy import Column from sqlalchemy import desc from sqlalchemy import func from sqlalchemy import Numeric from sqlalchemy import or_ from sqlalchemy import String from sqlalchemy.engine import Row from moneyhub.constants.constants import OrderDirection 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 RevenueByTransactionType(BaseModel): """Revenue by transaction type model.""" __tablename__ = 'revenue_by_transaction_type_dbt' account_id = Column(Numeric(12, 0), nullable=False, primary_key=True) artist_id = Column(Numeric(32, 0), nullable=False) product_id = Column(Numeric(38, 0), nullable=False) track_unique_id = Column(Numeric(32, 0), nullable=False) account_payee_currency = Column(String(50), nullable=False) contract_id = Column(Numeric(12, 0), nullable=False, primary_key=True) statement_period_id = Column(Numeric(38, 0), nullable=False, primary_key=True) subaccount_id = Column(Numeric(12, 0), nullable=True) transaction_type_id = Column(Numeric(38, 0), nullable=False) transaction_type_code = Column(String(4), nullable=True) transaction_type_desc = Column(String(180), nullable=False) transaction_type_group_id = Column(Numeric(38, 0), nullable=True, primary_key=True) transaction_type_group_name = Column(String(180), nullable=False) quantity = Column(Numeric(30, 0), nullable=False) net_revenue_payee_currency = Column(Numeric(36, 12), nullable=False) gross_revenue_payee_currency = Column(Numeric(36, 12), nullable=False) activity_period_id = Column(Numeric(38, 0), nullable=True) store_id = Column(Numeric(32, 0), nullable=True) country_code = Column(String(3), nullable=True) imprint_id = Column(Numeric(32, 0), nullable=True) project_id = Column(Numeric(32, 0), nullable=True) @classmethod def get_by_account_id( cls, 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, 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, subaccount_info: dict | None = None, project_id: int | None = None, ) -> list: """Get a list of revenue by transaction type for a specified account. Optionally filtered by contract and statement period range. Args: account_id (int): ID of an account artist_id (int): the id of an artist to filter by contract_id (int): Optional ID of a contract to filter by statement_period_id_start (int): Optional start of the period range statement_period_id_end (int): Optional end of the period range activity_period_id_start (int): Optional start of the activity period range activity_period_id_end (int): Optional 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): Search term to filter transaction types 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 subaccount_info (dict): Subaccount information project_id (int | None): project id to filter by Returns: list: list of revenue by transaction type """ filters = [(cls.account_id == account_id)] group_by = [ cls.transaction_type_id, cls.transaction_type_desc, cls.transaction_type_group_name, cls.account_payee_currency ] with_entities = [ cls.transaction_type_id, cls.transaction_type_desc, cls.transaction_type_group_name, cls.account_payee_currency, func.sum(cls.quantity).label('quantity'), func.sum(cls.net_revenue_payee_currency).label('net_revenue_payee_currency'), func.sum(cls.gross_revenue_payee_currency).label('gross_revenue_payee_currency') ] if artist_id: filters.append(cls.artist_id == artist_id) if contract_id: filters.append(cls.contract_id == contract_id) apply_filters( cls=cls, filters=filters, activity_period_id_start=activity_period_id_start, activity_period_id_end=activity_period_id_end, 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, store_ids=store_ids, subaccount_id=subaccount_id, track_unique_id=track_unique_id, transaction_type_ids=transaction_type_ids, ) if search_term: search_filter = cls.transaction_type_desc.ilike(f'%{search_term}%') if search_term.isnumeric(): search_filter = or_(search_filter, cls.transaction_type_id == search_term) filters.append(search_filter) order_direction = desc if order_dir == OrderDirection.DESC else asc apply_subaccount_revenue(subaccount_info, with_entities, cls) return cls.query \ .with_entities(*with_entities) \ .filter(*filters) \ .group_by(*group_by) \ .order_by(order_direction(order_by)) \ .all() @classmethod def get_transaction_types_by_account_id( cls, account_id: int, subaccount_id: int | None = None, ) -> list: """Get a list of revenue by transaction type for a specified account. Optionally filtered by subaccount. Args: account_id (int): ID of an account subaccount_id (int): Optional ID of a subaccount Returns: list: list of revenue by transaction type """ filters = [(cls.account_id == account_id)] with_entities = [ cls.transaction_type_id, cls.transaction_type_desc, cls.transaction_type_group_name, ] if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) return cls.query \ .with_entities(*with_entities) \ .filter(*filters) \ .order_by(asc(cls.transaction_type_desc)) \ .distinct() \ .all() @classmethod def get_transaction_types_totals_by_account_id( cls, 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, 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, subaccount_info: dict | None = None, project_id: int | None = None, ) -> Row: """GET total net and gross revenue for all transaction types for an account. 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 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 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 subaccount_info (dict): Subaccount information project_id (int | None): project id to filter by Returns: Row: Row containing the total net and gross revenue in the payee currency """ filters = [(cls.account_id == account_id)] with_entities = [ func.sum(cls.net_revenue_payee_currency).label('net_revenue_payee_currency'), func.sum(cls.gross_revenue_payee_currency).label('gross_revenue_payee_currency') ] if artist_id: filters.append(cls.artist_id == artist_id) if contract_id: filters.append(cls.contract_id == contract_id) if statement_period_id_start and statement_period_id_end: filters.append(cls.statement_period_id.between( statement_period_id_start, statement_period_id_end)) if activity_period_id_start and activity_period_id_end: filters.append(cls.activity_period_id.between( activity_period_id_start, activity_period_id_end)) if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if product_id: filters.append(cls.product_id == product_id) if project_id and hasattr(cls, 'project_id'): filters.append(cls.project_id == project_id) if track_unique_id: filters.append(cls.track_unique_id == track_unique_id) if store_ids: filters.append(cls.store_id.in_(store_ids)) if country_codes: filters.append(cls.country_code.in_(country_codes)) if imprint_ids: filters.append(cls.imprint_id.in_(imprint_ids)) if transaction_type_ids: filters.append(cls.transaction_type_id.in_(transaction_type_ids)) apply_subaccount_revenue(subaccount_info, with_entities, cls) return cls.query.with_entities(*with_entities).filter(*filters).one()