"""RevenueByImprint 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 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 RevenueByImprint(BaseModel): """Revenue By Imprint model.""" __tablename__ = 'revenue_by_imprint_dbt' imprintid = Column(Numeric(38, 0), nullable=False, primary_key=True) imprint = Column(String(16777216), nullable=True) account_id = Column(Numeric(12, 0), nullable=False, primary_key=True) subaccount_id = Column(Numeric(12, 0), nullable=True) artist_id = Column(Numeric(32, 0), nullable=False) contract_id = Column(Numeric(12, 0), nullable=True, primary_key=True) statement_period_id = Column(Numeric(38, 0), nullable=False, primary_key=True) account_payee_currency = Column(String(50), nullable=False) mechanical_deduction_amount_payee_currency = Column(Numeric(36, 12), nullable=False) publisher_admin_fee_payee_currency = Column(Numeric(36, 12), nullable=False) net_share_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(2), nullable=True) transaction_type_id = Column(Numeric(32, 0), nullable=True) @classmethod def get_by_account_id( cls, 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, subaccount_info: dict | None, ) -> tuple[list, int]: """GET list of revenue by imprint for a specified account ID and contract ID. Args: account_id (int): the id of an account artist_id (int): the id of an artist to filter by limit (int): how many entities to retrieve. offset (int): the offset (for pagination). 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): the id of a subaccount order_by (str): field to sort by 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 imprint_ids (list[int]): list of imprints to filter by transaction_type_ids (list[int]): list of transaction type IDs to filter by subaccount_info (dict): Subaccount information Returns: list: list of revenue by imprint """ filters = [(cls.account_id == account_id)] group_by = [ cls.imprintid, cls.imprint, cls.account_payee_currency, cls.account_id, cls.subaccount_id ] with_entities = [ cls.account_id, cls.imprintid, cls.imprint, cls.account_payee_currency, cls.subaccount_id, func.sum(cls.mechanical_deduction_amount_payee_currency).label( 'mechanical_deduction_amount_payee_currency'), func.sum(cls.publisher_admin_fee_payee_currency).label( 'publisher_admin_fee_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.count().over().label('total_records'), ] apply_subaccount_revenue(subaccount_info, with_entities, cls) if artist_id: filters.append(cls.artist_id == artist_id) if contract_id: filters.append(cls.contract_id == contract_id) with_entities.append(cls.contract_id) group_by.append(cls.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, statement_period_id_start=statement_period_id_start, statement_period_id_end=statement_period_id_end, store_ids=store_ids, subaccount_id=subaccount_id, transaction_type_ids=transaction_type_ids, ) if imprint_ids: filters.append(cls.imprintid.in_(imprint_ids)) if search_term: if search_term.isnumeric(): filters.append( or_(cls.imprint.ilike(f'%{search_term}%'), cls.imprintid == search_term)) else: filters.append(cls.imprint.ilike(f'%{search_term}%')) order_direction = desc if order_dir == OrderDirection.DESC else asc query = cls.query.with_entities(*with_entities).filter(*filters).group_by( *group_by).order_by(order_direction(order_by)) 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 @classmethod def get_imprints_by_account_id( cls, account_id: int, limit: int | None, search_term: str | None, subaccount_id: int | None, imprint_ids: list[int] | None ) -> list: """Get list of distinct imprints for a specified account. Args: account_id (int): the id of an account subaccount_id (int | None): the id of a subaccount to filter by limit (int | None): maximum number of imprints to return search_term (str | None): search term to filter imprints by name imprint_ids (list | None): imprint IDs to filter for Returns: list: list of distinct imprints with id, name, account_id, and subaccount_id """ filters = [cls.account_id == account_id] if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if search_term: if search_term.isnumeric(): filters.append(or_( cls.imprint.ilike(f'%{search_term}%'), cls.imprintid == search_term)) else: filters.append(cls.imprint.ilike(f'%{search_term}%')) if imprint_ids: filters.append(cls.imprintid.in_(imprint_ids)) with_entities = [ cls.imprintid, cls.imprint, cls.account_id, ] group_by = [cls.imprintid, cls.imprint, cls.account_id] query = cls.query.with_entities(*with_entities).filter(*filters).group_by( *group_by).order_by(cls.imprint) records = query.limit(limit).all() return records