"""RevenueByProject 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 RevenueByProject(BaseModel): """Revenue by project model.""" __tablename__ = 'revenue_by_project_dbt' project_id = Column(Numeric(32, 0), nullable=False, primary_key=True) project_name = Column(String(255), nullable=True) project_code = Column(String(50), nullable=True) project_artist = Column(String(255), nullable=True) account_id = Column(Numeric(12, 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) statement_period_id = Column(Numeric(38, 0), nullable=False, primary_key=True) 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) transaction_type_id = Column(Numeric(32, 0), nullable=True) account_payee_currency = Column(String(50), nullable=False) product_count = Column(Numeric(38, 0), nullable=False) net_share_payee_currency = Column(Numeric(36, 12), nullable=False) gross_revenue_payee_currency = Column(Numeric(36, 12), nullable=False) @classmethod def get_by_account_id( cls, 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, subaccount_info: dict | None, 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, ) -> tuple[list, int]: """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 ID filter subaccount_id (int | None): Optional subaccount ID filter statement_period_id_start (int | None): Optional statement period start statement_period_id_end (int | None): Optional statement period end order_by (str): Field to order by order_dir (OrderDirection): Sort direction search_term (str | None): Search by project fields subaccount_info (dict | None): Optional subaccount commission data activity_period_id_start (int | None): Optional activity period start activity_period_id_end (int | None): Optional activity period end 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 store_ids (list[int] | None): Optional store filter Returns: tuple[list, int]: grouped project rows and total grouped record count """ filters = [cls.account_id == account_id] group_by = [ cls.project_id, cls.project_name, cls.project_code, cls.project_artist, cls.product_count, cls.account_payee_currency, ] with_entities = [ cls.project_id, cls.project_name, cls.project_code, cls.project_artist, cls.product_count, 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.count().over().label('total_records'), ] apply_filters( cls=cls, filters=filters, contract_id=contract_id, subaccount_id=subaccount_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, country_codes=country_codes, imprint_ids=imprint_ids, transaction_type_ids=transaction_type_ids, store_ids=store_ids, ) if search_term: search_filters = [ cls.project_name.ilike(f'%{search_term}%'), cls.project_code.ilike(f'%{search_term}%'), cls.project_artist.ilike(f'%{search_term}%'), ] if search_term.isnumeric(): search_filters.append(cls.project_id == search_term) filters.append(or_(*search_filters)) apply_subaccount_revenue(subaccount_info, with_entities, cls) 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() total_records = int(records[0].total_records) if records else 0 return records, total_records @classmethod def get_projects_by_account_id( cls, account_id: int, limit: int, search_term: str | None, subaccount_id: int | None, ) -> list: """Get list of distinct projects 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): maximum number of projects to return search_term (str | None): search term to filter projects by code, name, or exact ID Returns: list: list of distinct projects """ filters = [cls.account_id == account_id, cls.project_code.isnot(None)] if subaccount_id: filters.append(cls.subaccount_id == subaccount_id) if search_term: search_filters = [ cls.project_code.ilike(f'%{search_term}%'), cls.project_name.ilike(f'%{search_term}%'), ] if search_term.isnumeric(): search_filters.append(cls.project_id == search_term) filters.append(or_(*search_filters)) with_entities = [ cls.project_id ] query = cls.query.with_entities(*with_entities) \ .filter(*filters) \ .distinct() \ .order_by(cls.project_name) records = query.limit(limit).all() return records