"""Snowflake connector."""

from collections.abc import Generator
from datetime import datetime
from decimal import Decimal
import math
from typing import Any
from typing import cast
from uuid import uuid4

from lambdacommon.common_config import logger
import pandas as pd
from pandas import concat
from pandas import DataFrame
from pandas import Series
from pandas import to_numeric
from snowflake import connector

import config
from config import DEV_ENVIRONMENT
from config import GROSS_REVENUE_COLUMN
from config import NET_REVENUE_COLUMN
from config import QA_ENVIRONMENT
from config import RevenueDisplayType
from src.connectors.ows_moneyhub import OwsMoneyhub
from src.exceptions import CustomReportException
from src.utils.constants import FeatureFlag
from src.utils.features import is_feature_enabled
from src.utils.types import DimensionMapping
from src.utils.types import SnowflakeQueryParams

ITERATOR_SIZE = 100000
MECHANICALS_HEADER = 'US Mechanicals'

# Mapping of filter names to their SQL column names
FILTER_COLUMN_MAPPING: dict[str, tuple[str, str]] = {
    'account_id': ('ACCOUNT_ID', 'single'),
    'contract_id': ('CONTRACT_ID', 'single'),
    'statement_period_ids': ('STATEMENT_PERIOD_ID', 'list'),
    'subaccount_id': ('SUBACCOUNT_ID', 'single'),
    'activity_period_ids': ('ACTIVITY_PERIOD_ID', 'list'),
    'artist_ids': ('ARTIST_ID', 'list'),
    'country_codes': ('COUNTRY_CODE', 'list'),
    'imprint_ids': ('IMPRINT_ID', 'list'),
    'product_ids': ('PRODUCT_ID', 'list'),
    'project_ids': ('PROJECT_ID', 'list'),
    'store_ids': ('STORE_ID', 'list'),
    'subaccount_ids': ('SUBACCOUNT_ID', 'list'),
    'recording_ids': ('SOUND_RECORDING_ID', 'list'),
    'track_unique_ids': ('TRACK_UNIQUE_ID', 'list'),
    'transaction_type_ids': ('TRANSACTION_TYPE_ID', 'list'),
}


def get_subaccount_info(subaccount_id: int) -> dict:
    """Get a subaccount by ID."""
    sql = """
    SELECT
        COMMISSIONOVERRIDE,
        SUBACCOUNT_SPLIT_TYPE
    FROM FACTS.{schema}.DIM_SUBACCOUNT
    WHERE SUBACCOUNTID = %(subaccount_id)s
    """.format(
        schema=QA_ENVIRONMENT if config.ENVIRONMENT == DEV_ENVIRONMENT else config.ENVIRONMENT
    )
    params = {
        'subaccount_id': subaccount_id,
    }

    return cast(dict, _query_one(sql, params))


def _query_one(query: str, params: dict) -> dict | None:
    """Execute query and fetch a single row from Snowflake.

    Args:
        query (str): SQL statement for extracting data from Snowflake.
        params: Parameters to pass to the query.

    Returns:
        dict | None: Single row as dict or None if no results.
    """
    with connector.connect(**config.SNOWFLAKE_DB_CONFIG) as connection:
        with connection.cursor(connector.DictCursor) as cursor:
            cursor.execute(query, params)
            row = cursor.fetchone()
            return row if row else None


def _build_where_condition_for_query(table_name: str, filters: list[str]) -> str:
    """Build a where condition for the SQL query for fetching report distribution data.

    Args:
        table_name (str): Name of the DB table to fetch data from
        filters (list): List of filters
    Returns:
        str: Where condition for SQL query
    """
    where = []
    for filter_name in filters:
        if filter_name not in FILTER_COLUMN_MAPPING:
            raise CustomReportException(f'Unhandled query filter: {filter_name}')

        column_name, filter_type = FILTER_COLUMN_MAPPING[filter_name]

        if filter_type == 'single':
            where.append(f'{table_name}.{column_name} = %({filter_name})s')
        elif filter_type == 'list':
            where.append(f'{table_name}.{column_name} IN (%({filter_name})s)')

    where_condition = 'WHERE ' + ' AND '.join(where) if len(where) else ''
    return where_condition


def _build_join_for_query(
    column_dimension: DimensionMapping, row_dimension: DimensionMapping
) -> list[str]:
    """Build the joins for a SQL query for fetching report distribution data.

    Args:
        column_dimension (dict): Column dimension to group by
        row_dimension (dict): Row dimension to group by
    Returns:
        list: SQL joins for the query
    """
    joins = []
    if 'join' in column_dimension:
        joins.append(column_dimension['join'])
    if 'join' in row_dimension:
        joins.append(row_dimension['join'])
    return joins


def _build_report_query(
    table_name: str,
    column_dimension: DimensionMapping,
    row_dimension: DimensionMapping,
    filters: list[str],
    revenue_display_type: RevenueDisplayType,
    subaccount_id: int | None = None,
) -> str:
    """Build a SQL query for fetching report distribution data.

    Args:
        table_name (str): Name of the DB table to fetch data from
        column_dimension (dict): Column dimension to group by
        row_dimension (dict): Row dimension to group by
        filters (list): List of filters
        revenue_display_type (RevenueDisplayType): revenue display type to determine column
        subaccount_id (int): Optional subaccount ID to filter by and determine revenue calculation
    Returns:
        str: SQL query
    """
    where_condition = _build_where_condition_for_query(table_name, filters)
    revenue_map = {
        'net': NET_REVENUE_COLUMN,
        'gross': GROSS_REVENUE_COLUMN,
    }

    joins = _build_join_for_query(column_dimension, row_dimension)

    mechanicals = (
        f"""
        COALESCE(SUM({table_name}.{config.MECHANICAL_COLUMN}), 0)
            + COALESCE(SUM({table_name}.{config.ADMIN_FEE_COLUMN}), 0)
            AS MECHANICALS,
        """
        if row_dimension.get('include_mechanicals', False)
        else ''
    )

    if subaccount_id:
        commission_override, split_type = get_subaccount_info(subaccount_id).values()
        revenue_column = f'{table_name}.{revenue_map[split_type.lower()]}'
        total_expression = f'SUM({revenue_column} * {commission_override})'
    else:
        total_expression = f'SUM({table_name}.{revenue_map[revenue_display_type]})'

    extra_columns = []
    if 'extra' in row_dimension:
        for extra in row_dimension['extra']:
            extra_columns.append(extra['column'])

            if 'join' in extra:
                joins.append(extra['join'])

    query = """
    SELECT
        {column_dimension_column} AS COLUMN_DIMENSION,
        {row_dimension_column} AS ROW_DIMENSION,
        {extra_columns}
        {table_name}.{currency_column} AS CURRENCY,
        {mechanicals}
        {total_expression} AS TOTAL
    FROM {table_name}
    {joins}
    {where_condition}
    GROUP BY
        {extra_columns}
        COLUMN_DIMENSION,
        ROW_DIMENSION,
        CURRENCY
    ORDER BY ROW_DIMENSION ASC
    """.format(
        column_dimension_column=column_dimension['column'],
        row_dimension_column=row_dimension['column'],
        joins=' '.join(joins),
        table_name=table_name,
        currency_column=config.CURRENCY_COLUMN,
        extra_columns=(','.join(extra_columns) + ',' if len(extra_columns) > 0 else ''),
        mechanicals=mechanicals,
        total_expression=total_expression,
        where_condition=where_condition,
    )

    return query


def _build_query_params(
    account_id: int,
    contract_id: int | None,
    statement_period_ids: list[int],
    subaccount_id: int | None,
    custom_filters: dict[str, list[int]] | None,
) -> SnowflakeQueryParams:
    """Build a dict of query parameters for getting report data.

    Args:
        account_id (int): Account to get sales for
        contract_id (int): Optional contract to filter by
        statement_period_ids (list): Statement periods to get sales for
        subaccount_id (int): Optional subaccount to filter by
        custom_filters (dict): Optional custom filters
    Returns
        dict: Key/value parameters for the query
    """
    params: SnowflakeQueryParams = {
        'account_id': account_id,
        'statement_period_ids': statement_period_ids,
    }
    if contract_id:
        params['contract_id'] = contract_id
    if subaccount_id:
        params['subaccount_id'] = subaccount_id

    if custom_filters:
        for key, value in custom_filters.items():
            if value is not None and len(value) > 0:
                params[key] = value

    return params


def _get_table_dimension_mappings(
    account_id: int, column_dimension: str, row_dimension: str, revenue_type: str
) -> tuple[str, DimensionMapping, DimensionMapping]:
    """Get the table and row/column dimension mappings for the query.

    Args:
        account_id (int): Account to get sales for
        column_dimension (str): Column dimension to group by
        row_dimension (str): Row dimension to group by
        revenue_type (str): Revenue type
    Returns:
        tuple: table name, column dimension mapping, row dimension mapping
    """
    table_name = None
    column_dimension_mapping = None
    row_dimension_mapping = None

    if revenue_type == config.REVENUE_TYPE_DISTRIBUTION:
        table_name = config.REPORT_TABLE_DISTRIBUTION
        column_dimension_mapping = config.REPORT_DIMENSION_MAP_DISTRIBUTION[column_dimension]
        row_dimension_mapping = config.REPORT_DIMENSION_MAP_DISTRIBUTION[row_dimension]
    elif revenue_type == config.REVENUE_TYPE_NEIGHBOURING_RIGHTS:
        table_name = config.REPORT_TABLE_NEIGHBOURING_RIGHTS
        column_dimension_mapping = config.REPORT_DIMENSION_MAP_NEIGHBOURING_RIGHTS[column_dimension]
        row_dimension_mapping = config.REPORT_DIMENSION_MAP_NEIGHBOURING_RIGHTS[row_dimension]
    else:
        raise CustomReportException(f'Unhandled revenue type: {revenue_type}')

    is_d3_account = is_distributor(account_id)['IS_DISTRIBUTOR'] == 'Y'
    if not is_d3_account:
        logger.info('Account not a D3 so filtering out columns')
        row_dimension_mapping = _filter_extras_for_non_d3(row_dimension_mapping)

    return table_name, column_dimension_mapping, row_dimension_mapping


def _pandas_query(query: str, params: SnowflakeQueryParams) -> DataFrame:
    """Run a query, returning a generator with dataframes and the total rows.

    Args:
        query (str): SQL statement for extracting data from Snowflake.
        params (dict): Parameters to pass along with the query.

    Returns:
        DataFrame: Generator containing DataFrame rows.
    """
    logger.info(f'Running query: {query}')
    with connector.connect(**config.SNOWFLAKE_DB_CONFIG) as connection:
        with connection.cursor(connector.DictCursor) as cursor:
            cursor.execute(query, params)

            df = cursor.fetch_pandas_all()
            logger.info(f'Got {len(df)} results!')
            return df


def _query(query: str, params: SnowflakeQueryParams) -> Generator[dict[str, Any], None, None]:
    """Extract data from Snowflake.

    Args:
        query (str): SQL statement for extracting data from Snowflake.
        params (SnowflakeQueryParams): Parameters bound to the SQL statement.

    Returns:
        Generator: Generator containing dict rows
    """
    logger.info(f'Running query: {query}')
    with connector.connect(**config.SNOWFLAKE_DB_CONFIG) as connection:
        with connection.cursor(connector.DictCursor) as cursor:
            cursor.execute(query, params)

            while True:
                logger.info('Fetching query results...')
                results = cursor.fetchmany(ITERATOR_SIZE)
                if not results:
                    logger.info('Got no results!')
                    break

                logger.info(f'Got {len(results)} results!')
                for result in results:
                    yield result


def _canon(x: object) -> str | None:
    """Collapse None/empty/whitespace to None; strip otherwise.

    Args:
        x (object): Value to normalize, non-None values are stringified and trimmed.

    Returns:
        str | None: Normalized string or None.
    """
    if x is None or (isinstance(x, float) and math.isnan(x)):
        return None
    s = str(x).replace('\xa0', ' ').strip()
    return s or None


def _should_include_mechanicals(
    row_dimension_mapping: DimensionMapping, account_id: int, subaccount_id: int | None
) -> bool:
    """Determine whether to include mechanicals in a report.

    Args:
        row_dimension_mapping (dict): Mapping for the row dimension
        account_id (int): ID of the account making the request
        subaccount_id (int): ID of the subaccount making the request
    Returns:
        bool: Whether to include mechanicals
    """
    if row_dimension_mapping.get('include_mechanicals', False) is False:
        return False

    if subaccount_id is not None:
        return False  # subaccount reports don't have mechanicals

    account_activity = OwsMoneyhub.get_account_activity(account_id)
    return bool(account_activity['mechanicals'])


def get_report_data(
    account_id: int,
    contract_id: int | None,
    statement_period_ids: list[int],
    column_dimension: str,
    row_dimension: str,
    revenue_type: str,
    revenue_display_type: RevenueDisplayType,
    subaccount_id: int | None = None,
    custom_filters: dict[str, list[int]] | None = None,
) -> DataFrame:
    """Get sales data for a report.

    Args:
        account_id (int): Account to get sales for
        contract_id (int): Optional contract to filter by
        statement_period_ids (list): Statement periods to get sales for
        column_dimension (str): Column dimension to group by
        row_dimension (str): Row dimension to group by
        revenue_type (str): Revenue type
        revenue_display_type (RevenueDisplayType): revenue display type to determine column
        subaccount_id (int): Optional subaccount to filter by
        custom_filters (dict): Optional custom filters
    Returns:
        DataFrame: Generator containing dict rows
    """
    logger.info(f'Using Pandas version {pd.__version__}')

    params = _build_query_params(
        account_id, contract_id, statement_period_ids, subaccount_id, custom_filters
    )
    table_name, column_dimension_mapping, row_dimension_mapping = _get_table_dimension_mappings(
        account_id, column_dimension, row_dimension, revenue_type
    )
    sql = _build_report_query(
        table_name,
        column_dimension_mapping,
        row_dimension_mapping,
        list(params.keys()),
        revenue_display_type,
        subaccount_id,
    )

    extra_fields = {}
    if 'extra' in row_dimension_mapping:
        for field in row_dimension_mapping['extra']:
            column_name = field['column'].split('.').pop()
            extra_fields[column_name] = field['header']

    row_dimension_header = row_dimension_mapping['header']

    if is_feature_enabled(FeatureFlag.CUSTOM_REPORTS_PANDAS_DATAFRAME, account_id):
        df = _fetch_dataframe_pandas(sql, params, extra_fields, row_dimension_mapping)
    else:
        df = _fetch_dataframe_legacy(sql, params, extra_fields, row_dimension_mapping)

    non_value_columns = ['CURRENCY', 'ROW_DIMENSION', MECHANICALS_HEADER, *extra_fields.keys()]
    column_dimension_columns = [column for column in df.columns if column not in non_value_columns]

    column_headers = []

    if df.empty:
        column_headers.append(row_dimension_header)
    else:
        df = df.rename(
            columns={
                'ROW_DIMENSION': row_dimension_header,
                **extra_fields,
            }
        ).set_index(row_dimension_header)

    column_headers += list(extra_fields.values())

    logger.info('Sorting columns...')

    # Sort the columns, then add extra fields first to headers
    if column_dimension == config.DimensionType.STATEMENT_PERIOD:
        column_dimension_columns.sort(key=lambda x: datetime.strptime(x, '%B %Y'))
    else:
        column_dimension_columns.sort()  # just alphabetical
    column_headers += column_dimension_columns
    column_headers.append(MECHANICALS_HEADER)
    column_headers.append('CURRENCY')
    df = df.reindex(columns=column_headers)

    logger.info('Converting numbers...')

    # Convert number columns to Pandas numeric type
    for column in column_dimension_columns:
        df[column] = to_numeric(df[column])
    df[MECHANICALS_HEADER] = to_numeric(df[MECHANICALS_HEADER])

    logger.info('Adding totals...')

    # Add a total for each column, using a unique key to avoid matching existing rows named 'Total'
    total_key = uuid4()
    df.loc[total_key] = df.sum(axis=0, numeric_only=True)
    df = df.rename({total_key: 'Total'})

    mechanicals = df.pop(MECHANICALS_HEADER)
    # Add a total for each row (with without mechanicals)
    if _should_include_mechanicals(row_dimension_mapping, account_id, subaccount_id):
        logger.info('Including subtotal and mechanicals column')
        df['Subtotal'] = df.sum(axis=1, numeric_only=True)
        df[MECHANICALS_HEADER] = mechanicals
        df['Total'] = df[['Subtotal', MECHANICALS_HEADER]].sum(axis=1, numeric_only=True)
    else:
        df['Total'] = df.sum(axis=1, numeric_only=True)

    # Move the currency field to the end
    df['CURRENCY'] = df.pop('CURRENCY')

    return df


def _fetch_dataframe_legacy(
    sql: str,
    params: SnowflakeQueryParams,
    extra_fields: dict[str, str],
    row_dimension_mapping: DimensionMapping,
) -> DataFrame:
    """Fetch the data into a dataframe based the legacy request method.

    Args:
        sql (str): SQL query to run
        params (dict): Parameters for the SQL query
        extra_fields (dict): Extra fields included in the mapping
        row_dimension_mapping (dict): Mapping for the row dimension
    Returns:
        DataFrame: Dataframe containing the data from the request
    """
    column_dimension_columns = []

    formatted_data = {}
    for item in _query(sql, params):
        column_value = _canon(item.get('COLUMN_DIMENSION'))
        row_value = _canon(item.get('ROW_DIMENSION'))
        currency = _canon(item.get('CURRENCY'))
        mechanicals = Decimal(item.get('MECHANICALS') or 0)
        revenue = Decimal(item.get('TOTAL') or 0)

        if column_value and column_value not in column_dimension_columns:
            column_dimension_columns.append(column_value)

        row_missing = row_value is None
        row_key: tuple[str | None, ...] = (None,)
        if not row_missing:
            row_key = (row_value,)
            for key in extra_fields.keys():
                row_key += (_canon(item.get(key)),)

        if row_key not in formatted_data:
            if row_missing and 'default' in row_dimension_mapping:
                formatted_data[row_key] = Series(
                    [row_dimension_mapping['default'], currency, 0],
                    index=['ROW_DIMENSION', 'CURRENCY', MECHANICALS_HEADER],
                )
            else:
                values: list[str | int | None] = [row_value, currency]
                indices: list[str] = ['ROW_DIMENSION', 'CURRENCY']

                for column, _ in extra_fields.items():
                    values.append(_canon(item.get(column)))
                    indices.append(column)

                values.append(0)
                indices.append(MECHANICALS_HEADER)

                formatted_data[row_key] = Series(values, indices)

        if column_value in formatted_data[row_key]:
            formatted_data[row_key][column_value] += revenue
        else:
            formatted_data[row_key] = concat(
                [formatted_data[row_key], Series([revenue], index=[column_value])]
            )

        formatted_data[row_key][MECHANICALS_HEADER] += mechanicals

        if 'CURRENCY' not in formatted_data[row_key] and currency is not None:
            formatted_data[row_key] = concat(
                [formatted_data[row_key], Series([currency], index=['CURRENCY'])]
            )

    logger.info('Creating dataframe with formatted data...')

    # Create DataFrame with Headers
    df = DataFrame(formatted_data.values())

    return df


def _fetch_dataframe_pandas(
    sql: str,
    params: SnowflakeQueryParams,
    extra_fields: dict[str, str],
    row_dimension_mapping: DimensionMapping,
) -> DataFrame:
    """Fetch the data into a dataframe based on a Pandas query.

    Args:
        sql (str): SQL query to run
        params (dict): Parameters for the SQL query
        extra_fields (dict): Extra fields included in the mapping
        row_dimension_mapping (dict): Mapping for the row dimension
    Returns:
        DataFrame: Dataframe containing the data from the request
    """
    df = (
        _pandas_query(sql, params)
        .map(_canon)  # trim/conform values
        .fillna({field: '' for field in extra_fields.keys()})  # make extra fields empty strings
    )

    # Fill missing values (needed for index and aggregation)
    df['ROW_DIMENSION'] = df['ROW_DIMENSION'].fillna('')
    if 'MECHANICALS' not in df:
        df['MECHANICALS'] = 0

    # Enforce numeric types for pivot values
    df['TOTAL'] = to_numeric(df['TOTAL'])
    df['MECHANICALS'] = to_numeric(df['MECHANICALS'])

    # Create pivot tables (summing up mechanicals separately)
    index = [
        'ROW_DIMENSION',
        'CURRENCY',
        *extra_fields.keys(),
    ]
    mechanicals_df = df.pivot_table(index=index, values='MECHANICALS', aggfunc='sum')
    df = df.pivot_table(index=index, columns='COLUMN_DIMENSION', values='TOTAL', aggfunc='sum')
    df[MECHANICALS_HEADER] = mechanicals_df['MECHANICALS'] if not mechanicals_df.empty else 0.0
    df[MECHANICALS_HEADER] = df[MECHANICALS_HEADER].fillna(0.0)

    df = df.reset_index()  # removes the pivot grouping

    # Add default values
    if 'default' in row_dimension_mapping:
        df['ROW_DIMENSION'] = df['ROW_DIMENSION'].replace('', row_dimension_mapping['default'])

    return df


def _build_report_query_financial_detail(
    row_dimension: DimensionMapping, filters: list[str], table_name: str
) -> str:
    """Build a SQL query for fetching report data.

    Args:
        row_dimension (dict): Row dimension to group by
        filters (list): List of filters
        table_name (str): Name of the DB table to fetch data from
    Returns:
        str: SQL query
    """
    where = []
    for column in filters:
        if column == 'account_id':
            where.append(f'{table_name}.ACCOUNT_ID = %(account_id)s')
        elif column == 'contract_id':
            where.append(f'{table_name}.CONTRACT_ID = %(contract_id)s')
        elif column == 'statement_period_ids':
            where.append(f'{table_name}.STATEMENT_PERIOD_ID IN (%(statement_period_ids)s)')  # noqa: E501
        else:
            raise CustomReportException(f'Unhandled query filter: {column}')
    where_condition = 'WHERE ' + ' AND '.join(where) if len(where) else ''

    joins = []
    if 'join' in row_dimension:
        joins.append(row_dimension['join'])

    extra_columns = []
    if 'extra' in row_dimension:
        for extra in row_dimension['extra']:
            extra_columns.append(extra['column'])

            if 'join' in extra:
                joins.append(extra['join'])

    query = """
    SELECT
        {row_dimension_column} AS ROW_DIMENSION,
        {extra_columns}
        SUM({table_name}.GROSS_REVENUE_PAYEE_CURRENCY) AS PRE_WHT_AMOUNT,
        SUM({table_name}.WITHHOLDING_TAX_PAYEE_CURRENCY) AS WHT_AMOUNT,
        SUM({table_name}.GROSS_REVENUE_AFTER_WITHHOLDING_TAX_PAYEE_CURRENCY)
            AS GROSS_REVENUE,
        SUM({table_name}.GROSS_REVENUE_AFTER_WITHHOLDING_TAX_PAYEE_CURRENCY
                    - {table_name}.NET_SHARE_PAYEE_CURRENCY) AS COMMISSION,
        SUM({table_name}.NET_SHARE_PAYEE_CURRENCY) AS NET_REVENUE,
        {table_name}.{currency_column} AS CURRENCY
    FROM {table_name}
    {joins}
    {where_condition}
    GROUP BY
        {extra_columns}
        ROW_DIMENSION,
        CURRENCY
    ORDER BY ROW_DIMENSION ASC
    """.format(
        row_dimension_column=row_dimension['column'],
        extra_columns=(','.join(extra_columns) + ',' if len(extra_columns) > 0 else ''),
        joins=' '.join(joins),
        table_name=table_name,
        currency_column=config.CURRENCY_COLUMN,
        where_condition=where_condition,
    )

    return query


def get_report_data_financial_detail(
    account_id: int,
    contract_id: int | None,
    statement_period_ids: list[int],
    row_dimension: str,
    revenue_type: str,
) -> DataFrame:
    """Get sales data for a report.

    Args:
        account_id (int): Account to get sales for
        contract_id (int): Optional contract to filter by
        statement_period_ids (list): Statement periods to get sales for
        row_dimension (str): Row dimension to group by
        revenue_type (str): Revenue type
    Returns:
        Generator: Generator containing dict rows
    """
    table_name: str | None = None
    row_dimension_mapping: DimensionMapping | None = None
    params: SnowflakeQueryParams = {
        'account_id': account_id,
        'statement_period_ids': statement_period_ids,
    }
    if revenue_type == config.REVENUE_TYPE_NEIGHBOURING_RIGHTS:
        table_name = config.REPORT_TABLE_NEIGHBOURING_RIGHTS
        row_dimension_mapping = config.REPORT_DIMENSION_MAP_NEIGHBOURING_RIGHTS[row_dimension]
    if revenue_type == config.REVENUE_TYPE_DISTRIBUTION:
        table_name = config.REPORT_TABLE_DISTRIBUTION
        row_dimension_mapping = config.REPORT_DIMENSION_MAP_DISTRIBUTION[row_dimension]
    if contract_id:
        params['contract_id'] = contract_id

    if table_name is None or row_dimension_mapping is None:
        raise CustomReportException(f'Unhandled revenue type: {revenue_type}')

    row_dimension_header = row_dimension_mapping['header']
    column_dimension_columns = [
        'PRE_WHT_AMOUNT',
        'WHT_AMOUNT',
        'GROSS_REVENUE',
        'COMMISSION',
        'NET_REVENUE',
    ]

    sql = _build_report_query_financial_detail(
        row_dimension_mapping, list(params.keys()), table_name
    )
    extra_fields = {}
    if 'extra' in row_dimension_mapping:
        for field in row_dimension_mapping['extra']:
            column_name = field['column'].split('.').pop()
            extra_fields[column_name] = field['header']

    if not is_feature_enabled(FeatureFlag.CUSTOM_REPORTS_PANDAS_DATAFRAME, account_id):
        return _legacy_get_report_data_financial_detail(
            sql,
            params,
            extra_fields,
            row_dimension_header,
            column_dimension_columns,
            row_dimension_mapping,
        )

    df = (
        _pandas_query(sql, params)
        .map(_canon)  # trim/conform values
        .fillna({field: '' for field in extra_fields.keys()})  # make extra fields empty strings
    )

    column_headers = []
    if df.empty:
        column_headers.append(row_dimension_header)
    else:
        # Fill missing values (needed for index and aggregation)
        if 'default' in row_dimension_mapping:
            df['ROW_DIMENSION'] = df['ROW_DIMENSION'].fillna(row_dimension_mapping['default'])
        else:
            df['ROW_DIMENSION'] = df['ROW_DIMENSION'].fillna('')

        df = df.rename(
            columns={
                'ROW_DIMENSION': row_dimension_header,
                **extra_fields,
            }
        ).set_index(row_dimension_header)

    column_headers += list(extra_fields.values()) + column_dimension_columns
    column_headers.append('CURRENCY')
    df = df.reindex(columns=column_headers)

    # Convert number columns to Pandas numeric type
    for column in column_dimension_columns:
        df[column] = to_numeric(df[column])

    # Add a total for each column
    df.loc['Total'] = df.sum(axis=0, numeric_only=True)
    df['CURRENCY'] = df.pop('CURRENCY')

    return df


def _legacy_get_report_data_financial_detail(
    sql: str,
    params: dict,
    extra_fields: dict,
    row_dimension_header: str,
    column_dimension_columns: list[str],
    row_dimension_mapping: DimensionMapping,
) -> DataFrame:
    """Get report data using the old for loop method.

    Args:
        sql (str): SQL query to send to Snowflake
        params (dict): Parameters for the SQL query
        extra_fields (dict): Extra fields that are included in the mapping
        row_dimension_header (str): Name of the row dimension
        column_dimension_columns (list): List of columns in the column dimension
        row_dimension_mapping (DimensionMapping): Mapping for the row dimension
    Returns:
        DataFrame: Formatted financial data
    """
    formatted_data = {}

    for item in _query(sql, params):
        row_value = item['ROW_DIMENSION']
        currency = item['CURRENCY']
        row_key: tuple[Any, ...] = (item['ROW_DIMENSION'],)
        for key in extra_fields.keys():
            row_key += (item[key],)

        if row_key not in formatted_data:
            if not row_value and 'default' in row_dimension_mapping:
                # Skip over the extra fields if using the default value
                formatted_data[row_key] = Series(
                    [row_dimension_mapping['default']], [row_dimension_header]
                )
            else:
                values = [row_value, currency]
                indices = [row_dimension_header, 'CURRENCY']

                for column, header in extra_fields.items():
                    values.append(item[column])
                    indices.append(header)

                formatted_data[row_key] = Series(values, indices)
        for column in column_dimension_columns:
            if column in formatted_data[row_key]:
                formatted_data[row_key][column] += item[column]
            else:
                formatted_data[row_key] = concat(
                    [formatted_data[row_key], Series(item[column], [column])]
                )

    # Create DataFrame with Headers
    df = DataFrame(formatted_data.values())
    column_headers = []

    if df.empty:
        column_headers.append(row_dimension_header)
    else:
        df = df.set_index(row_dimension_header)

    column_headers += list(extra_fields.values()) + column_dimension_columns
    column_headers.append('CURRENCY')
    df = df.reindex(columns=column_headers)

    # Convert number columns to Pandas numeric type
    for column in column_dimension_columns:
        df[column] = to_numeric(df[column])

    # Add a total for each column
    df.loc['Total'] = df.sum(axis=0, numeric_only=True)
    df['CURRENCY'] = df.pop('CURRENCY')

    return df


def is_distributor(account_id: int) -> dict[str, Any]:
    """Check if account is a distributor."""
    sql = """
        SELECT CASE WHEN is_distributor = 'Y' THEN 'Y' ELSE 'N' END AS IS_DISTRIBUTOR
        FROM FACTS.{schema}.VENDOR
        WHERE VENDOR_ID = %(vendor_id)s
        LIMIT 1
    """.format(schema=config.DB_SCHEMA)
    gen = _query(sql, {'vendor_id': account_id})
    row = next(gen, None)
    if row is None:
        logger.warning('is_distributor: no row for account_id=%s; defaulting to N', account_id)
        return {'IS_DISTRIBUTOR': 'N'}
    return row


def get_subaccount_name(subaccount_id: int) -> str | None:
    """Get subaccount name by subaccount_id.

    Args:
        subaccount_id (int): The subaccount ID to look up
    Returns:
        str | None: The subaccount name or None if not found
    """
    sql = """
        SELECT SUBACCOUNT_NAME
        FROM ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.SUBACCOUNT
        WHERE SUBACCOUNT_ID = %(subaccount_id)s
        LIMIT 1
    """
    gen = _query(sql, {'subaccount_id': subaccount_id})
    row = next(gen, None)
    if row is None:
        logger.warning('get_subaccount_name: no row for subaccount_id=%s', subaccount_id)
        return None
    return row.get('SUBACCOUNT_NAME')


_NON_D3_ACCOUNT_EXTRA_EXCLUDE = {'SUBACCOUNT'}


def get_statement_periods_parsed(statement_period_ids: list[int]) -> list[tuple[int, int]] | None:
    """Get statement period month and year by statement period IDs.

    Args:
        statement_period_ids (list[int]): List of statement period IDs
    Returns:
        list[tuple]: List of statement period month and year tuples
    """
    sql = """
        SELECT STATEMENT_MONTH, STATEMENT_YEAR
        FROM
            ORCHARD_APP_REPORTING_V2.{schema}_ROYALTY_ACCOUNTING_ROYALTY_ACCOUNTING.STATEMENT_PERIOD
        WHERE STATEMENT_PERIOD_ID IN (%(statement_period_ids)s)
    """.format(schema=config.DB_SCHEMA)
    gen = _query(sql, {'statement_period_ids': statement_period_ids})
    parsed_statement_periods = []
    for row in gen:
        month = row.get('STATEMENT_MONTH')
        year = row.get('STATEMENT_YEAR')
        if month and year:
            parsed_statement_periods.append((year, month))
    if not parsed_statement_periods:
        logger.warning(
            'get_statement_period_names: no rows for statement_period_ids=%s', statement_period_ids
        )
        return None
    return parsed_statement_periods


def _filter_extras_for_non_d3(mapping: DimensionMapping) -> DimensionMapping:
    if 'extra' not in mapping:
        return mapping

    filtered_extras = [
        extra
        for extra in mapping['extra']
        if extra.get('header', '').strip().upper() not in _NON_D3_ACCOUNT_EXTRA_EXCLUDE
    ]

    return {**mapping, 'extra': filtered_extras}