import warnings
from datetime import datetime

import dagster as dg
from dagster import EnvVar
from dagster_snowflake import SnowflakeResource

from src.assets.snowflake.dim.user.assets import dim_user
from src.assets.snowflake.raw_table.frontend.app_audio_actions.assets import app_audio_actions
from src.assets.snowflake.raw_table.frontend.app_event.assets import app_event
from src.assets.snowflake.raw_table.frontend.web_audio_player_actions.assets import (
    web_audio_player_actions,
)
from src.assets.snowflake.raw_table.frontend.web_user_event.assets import web_user_event
from src.utils.automation_conditions import hourly_cron_with_eager_historical_backfill_condition
from src.utils.snowflake.constants import TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H, Group, PartitionExpr, Team, Warehouse
from src.utils.snowflake.query import JinjaSQLFormatter, PythonStringSQLFormatter

warnings.filterwarnings("ignore", category=dg.BetaWarning)

# Staging asset start dates
WEB_PLAY_START_DATE = datetime.strptime('2024-06-01', '%Y-%m-%d')
IOS_PLAY_START_DATE = datetime.strptime('2025-06-17', '%Y-%m-%d')
IOS_LEGACY_PLAY_START_DATE = datetime.strptime('2024-07-23', '%Y-%m-%d')
ANDROID_PLAY_START_DATE = datetime.strptime('2024-12-03', '%Y-%m-%d')
FACT_PLAY_START_DATE = min(WEB_PLAY_START_DATE, IOS_LEGACY_PLAY_START_DATE, IOS_PLAY_START_DATE, ANDROID_PLAY_START_DATE)

STG_WEB_PLAY_TABLE_NAME = "STG_WEB_PLAY"
STG_IOS_PLAY_TABLE_NAME = "STG_IOS_PLAY"
STG_IOS_LEGACY_PLAY_TABLE_NAME = "STG_IOS_LEGACY_PLAY"
STG_ANDROID_PLAY_TABLE_NAME = "STG_ANDROID_PLAY"
FACT_PLAY_TABLE_NAME = "FACT_PLAY_V2"

class StgWebPlayConfig(dg.Config):
    web_user_event_dedup_table_name: str = "TEMP_WEB_USER_EVENT_DEDUP"
    web_audio_player_actions_dedup_table_name: str = "TEMP_WEB_AUDIO_PLAYER_ACTIONS_DEDUP"
    web_events_per_song_session_table_name: str = "TEMP_WEB_EVENTS_PER_SONG_SESSION"
    web_unique_play_segments_table_name: str = "TEMP_WEB_UNIQUE_PLAY_SEGMENTS"
    web_play_sessions_table_name: str = "TEMP_WEB_PLAY_SESSIONS"
    web_stg_play_table_name: str = STG_WEB_PLAY_TABLE_NAME
    warehouse: str = Warehouse.FACT_PLAY_BACKFILL_SMALL.value

class StgIosPlayConfig(dg.Config):
    ios_omniplayer_events_dedup_table_name: str = "TEMP_IOS_OMNIPLAYER_EVENTS_DEDUP"
    ios_unique_play_segments_table_name: str = "TEMP_IOS_UNIQUE_PLAY_SEGMENTS"
    ios_play_sessions_table_name: str = "TEMP_IOS_PLAY_SESSIONS"
    ios_stg_play_table_name: str = STG_IOS_PLAY_TABLE_NAME
    warehouse: str = Warehouse.FACT_PLAY_BACKFILL_SMALL.value

class StgIosLegacyPlayConfig(dg.Config):
    ios_legacy_audio_actions_dedup_table_name: str = "TEMP_IOS_LEGACY_AUDIO_ACTIONS_DEDUP"
    ios_legacy_ordered_actions_table_name: str = "TEMP_IOS_LEGACY_ORDERED_ACTIONS"
    ios_legacy_play_durations_table_name: str = "TEMP_IOS_LEGACY_PLAY_DURATIONS"
    ios_legacy_play_sessions_table_name: str = "TEMP_IOS_LEGACY_PLAY_SESSIONS"
    ios_legacy_stg_play_table_name: str = STG_IOS_LEGACY_PLAY_TABLE_NAME
    warehouse: str = Warehouse.FACT_PLAY_BACKFILL_SMALL.value

class StgAndroidPlayConfig(dg.Config):
    android_audio_player_events_dedup_table_name: str = "TEMP_ANDROID_AUDIO_PLAYER_EVENTS_DEDUP"
    android_unique_play_segments_table_name: str = "TEMP_ANDROID_UNIQUE_PLAY_SEGMENTS"
    android_play_sessions_table_name: str = "TEMP_ANDROID_PLAY_SESSIONS"
    android_stg_play_table_name: str = STG_ANDROID_PLAY_TABLE_NAME
    warehouse: str = Warehouse.FACT_PLAY_BACKFILL_SMALL.value

class FactPlayConfig(dg.Config):
    web_stg_play_table_name: str = STG_WEB_PLAY_TABLE_NAME
    ios_stg_play_table_name: str = STG_IOS_PLAY_TABLE_NAME
    ios_legacy_stg_play_table_name: str = STG_IOS_LEGACY_PLAY_TABLE_NAME
    android_stg_play_table_name: str = STG_ANDROID_PLAY_TABLE_NAME
    fact_play_table_name: str = FACT_PLAY_TABLE_NAME
    warehouse: str = Warehouse.FACT_PLAY_BACKFILL_SMALL.value


@dg.asset(
    name="stg_web_play",
    description="Staging table for web play events. Processes WEB_USER_EVENT and WEB_AUDIO_PLAYER_ACTIONS into structured play sessions.",
    group_name=Group.LISTENING.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=WEB_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(web_user_event, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=-1, end_offset=1)),
        dg.AssetDep(web_audio_player_actions, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=-1, end_offset=1)),
        dg.AssetDep(["PROD", "dim_clip"]),
        dg.AssetDep(dim_user),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "data_start_date": WEB_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_web_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: StgWebPlayConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Because song sessions can span multiple hour partitions, we use a buffered fetch window
    # then filter down to the actual partition window. We use +1 to account for sessions that end in the next hour,
    # and -1 to correctly exclude sessions that actually start in the previous hour.
    is_multi_partition_range = context.has_partition_key_range
    fetch_window = context.asset_partitions_time_window_for_input(web_audio_player_actions.key.to_user_string())
    fetch_start_ts = fetch_window.start
    fetch_end_ts = fetch_window.end

    fetch_params = {
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "web_user_event_dedup_table_name": config.web_user_event_dedup_table_name,
        "web_audio_player_actions_dedup_table_name": config.web_audio_player_actions_dedup_table_name,
        "web_events_per_song_session_table_name": config.web_events_per_song_session_table_name,
        "web_unique_play_segments_table_name": config.web_unique_play_segments_table_name,
        "web_play_sessions_table_name": config.web_play_sessions_table_name,
        "web_stg_play_table_name": config.web_stg_play_table_name,
    }

    logger.info(f"Processing target partition: {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Create temporary tables with deduplicated data
        logger.info(f"Creating temporary dedup table {config.web_user_event_dedup_table_name}...")
        dedup_web_user_event_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_user_event_deduplicated.sql", params=fetch_params, logger=logger)
        cursor.execute(dedup_web_user_event_query)

        logger.info(f"Creating temporary dedup table {config.web_audio_player_actions_dedup_table_name}...")
        dedup_web_audio_player_actions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_audio_player_actions_deduplicated.sql", params=fetch_params, logger=logger)
        cursor.execute(dedup_web_audio_player_actions_query)

        # 2. Count events per song session (for filtering)
        logger.info(f"Creating temporary table {config.web_events_per_song_session_table_name} with event counts...")
        events_per_session_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_play_events_per_session.sql", params=fetch_params, logger=logger)
        cursor.execute(events_per_session_query)

        # 3. Create temporary table with unique play segments
        logger.info(f"Creating temporary table {config.web_unique_play_segments_table_name} with unique play segments...")
        unique_segments_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_unique_play_segments.sql", params=fetch_params, logger=logger)
        cursor.execute(unique_segments_query)

        # 4. Create temporary table with aggregated sessions (with event filtering)
        logger.info(f"Creating temporary table {config.web_play_sessions_table_name} with aggregated play sessions...")
        sessions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_play_sessions.sql", params=fetch_params, logger=logger)
        cursor.execute(sessions_query)

        # 5. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.web_stg_play_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={
            **fetch_params,
            "delete_partition_table_name": config.web_stg_play_table_name,
        }, logger=logger)
        cursor.execute(delete_query)

        # 6. Insert new web play data into the target table
        logger.info(f"Inserting web play data into {config.web_stg_play_table_name}...")
        insert_web_play_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_web_play.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_web_play_query)
        web_rows = cursor.rowcount

    logger.info(f"Successfully processed partition. Inserted {web_rows} rows")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.web_stg_play_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": web_rows if not is_multi_partition_range else 0,
        },
    )

@dg.asset(
    name="stg_ios_play",
    description="Staging table for iOS play events. Processes APP_EVENT omniplayer category into structured play sessions.",
    group_name=Group.LISTENING.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=IOS_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(app_event, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=-1, end_offset=1)),
        dg.AssetDep(["PROD", "dim_clip"]),
        dg.AssetDep(dim_user),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "data_start_date": IOS_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_ios_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: StgIosPlayConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Get input data window (0 to +1 hour from current partition)
    is_multi_partition_range = context.has_partition_key_range
    fetch_window = context.asset_partitions_time_window_for_input(app_event.key.to_user_string())
    fetch_start_ts = fetch_window.start
    fetch_end_ts = fetch_window.end

    fetch_params = {
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "ios_omniplayer_events_dedup_table_name": config.ios_omniplayer_events_dedup_table_name,
        "ios_unique_play_segments_table_name": config.ios_unique_play_segments_table_name,
        "ios_play_sessions_table_name": config.ios_play_sessions_table_name,
        "ios_stg_play_table_name": config.ios_stg_play_table_name,
    }

    logger.info(f"Processing target partition: {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Create temporary table with deduplicated iOS omniplayer events
        logger.info(f"Creating temporary table {config.ios_omniplayer_events_dedup_table_name} with deduplicated iOS omniplayer events...")
        dedup_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_play_events_deduplicated.sql", params=fetch_params, logger=logger)
        cursor.execute(dedup_query)

        # 2. Create temporary table with unique play segments
        logger.info(f"Creating temporary table {config.ios_unique_play_segments_table_name} with unique play segments...")
        unique_segments_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_unique_play_segments.sql", params=fetch_params, logger=logger)
        cursor.execute(unique_segments_query)

        # 3. Create temporary table with aggregated sessions
        logger.info(f"Creating temporary table {config.ios_play_sessions_table_name} with aggregated play sessions...")
        sessions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_play_sessions.sql", params=fetch_params, logger=logger)
        cursor.execute(sessions_query)

        # 4. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.ios_stg_play_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={
            **fetch_params,
            "delete_partition_table_name": config.ios_stg_play_table_name,
        }, logger=logger)
        cursor.execute(delete_query)

        # 5. Insert new iOS play data into the target table
        logger.info(f"Inserting iOS play data into {config.ios_stg_play_table_name}...")
        insert_ios_play_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_play.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_ios_play_query)
        ios_rows = cursor.rowcount

    logger.info(f"Successfully processed partition. Inserted {ios_rows} rows")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.ios_stg_play_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": ios_rows if not is_multi_partition_range else 0,
        },
    )

@dg.asset(
    name="stg_ios_legacy_play",
    description="Staging table for iOS legacy play events. Processes APP_AUDIO_ACTIONS into structured play sessions.",
    group_name=Group.LISTENING.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=IOS_LEGACY_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(app_audio_actions, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=-1, end_offset=1)),
        dg.AssetDep(["PROD", "dim_clip"]),
        dg.AssetDep(dim_user),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "data_start_date": IOS_LEGACY_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_ios_legacy_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: StgIosLegacyPlayConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Get input data window (0 to +1 hour from current partition)
    is_multi_partition_range = context.has_partition_key_range
    fetch_window = context.asset_partitions_time_window_for_input(app_audio_actions.key.to_user_string())
    fetch_start_ts = fetch_window.start
    fetch_end_ts = fetch_window.end

    fetch_params = {
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "ios_legacy_audio_actions_dedup_table_name": config.ios_legacy_audio_actions_dedup_table_name,
        "ios_legacy_ordered_actions_table_name": config.ios_legacy_ordered_actions_table_name,
        "ios_legacy_play_durations_table_name": config.ios_legacy_play_durations_table_name,
        "ios_legacy_play_sessions_table_name": config.ios_legacy_play_sessions_table_name,
        "ios_legacy_stg_play_table_name": config.ios_legacy_stg_play_table_name,
    }

    logger.info(f"Processing target partition: {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Create temporary table with deduplicated iOS legacy audio actions
        logger.info(f"Creating temporary table {config.ios_legacy_audio_actions_dedup_table_name} with deduplicated iOS legacy audio actions...")
        dedup_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_legacy_play_events_deduplicated.sql", params=fetch_params, logger=logger)
        cursor.execute(dedup_query)

        # 2. Create temporary table with ordered actions (using LEAD window function)
        logger.info(f"Creating temporary table {config.ios_legacy_ordered_actions_table_name} with ordered actions and play durations...")
        ordered_actions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_legacy_ordered_actions.sql", params=fetch_params, logger=logger)
        cursor.execute(ordered_actions_query)

        # 3. Create temporary table with play durations
        logger.info(f"Creating temporary table {config.ios_legacy_play_durations_table_name} with play durations...")
        play_durations_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_legacy_play_durations.sql", params=fetch_params, logger=logger)
        cursor.execute(play_durations_query)

        # 4. Create temporary table with aggregated sessions
        logger.info(f"Creating temporary table {config.ios_legacy_play_sessions_table_name} with aggregated play sessions...")
        sessions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_legacy_play_sessions.sql", params=fetch_params, logger=logger)
        cursor.execute(sessions_query)

        # 5. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.ios_legacy_stg_play_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={
            **fetch_params,
            "delete_partition_table_name": config.ios_legacy_stg_play_table_name,
        }, logger=logger)
        cursor.execute(delete_query)

        # 6. Insert new iOS legacy play data into the target table
        logger.info(f"Inserting iOS legacy play data into {config.ios_legacy_stg_play_table_name}...")
        insert_ios_legacy_play_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_ios_legacy_play.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_ios_legacy_play_query)
        ios_legacy_rows = cursor.rowcount

    logger.info(f"Successfully processed partition. Inserted {ios_legacy_rows} rows")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.ios_legacy_stg_play_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": ios_legacy_rows if not is_multi_partition_range else 0,
        },
    )

@dg.asset(
    name="stg_android_play",
    description="Staging table for Android play events. Processes APP_EVENT audio_player category into structured play sessions.",
    group_name=Group.LISTENING.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=ANDROID_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(app_event, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=-1, end_offset=1)),
        dg.AssetDep(["PROD", "dim_clip"]),
        dg.AssetDep(dim_user),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "data_start_date": ANDROID_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_android_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: StgAndroidPlayConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Get input data window (0 to +1 hour from current partition)
    is_multi_partition_range = context.has_partition_key_range
    fetch_window = context.asset_partitions_time_window_for_input(app_event.key.to_user_string())
    fetch_start_ts = fetch_window.start
    fetch_end_ts = fetch_window.end

    fetch_params = {
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "android_audio_player_events_dedup_table_name": config.android_audio_player_events_dedup_table_name,
        "android_unique_play_segments_table_name": config.android_unique_play_segments_table_name,
        "android_play_sessions_table_name": config.android_play_sessions_table_name,
        "android_stg_play_table_name": config.android_stg_play_table_name,
    }

    logger.info(f"Processing target partition: {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Create temporary table with deduplicated Android audio player events
        logger.info(f"Creating temporary table {config.android_audio_player_events_dedup_table_name} with deduplicated Android audio player events...")
        dedup_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_android_play_events_deduplicated.sql", params=fetch_params, logger=logger)
        cursor.execute(dedup_query)

        # 2. Create temporary table with unique play segments
        logger.info(f"Creating temporary table {config.android_unique_play_segments_table_name} with unique play segments...")
        unique_segments_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_android_unique_play_segments.sql", params=fetch_params, logger=logger)
        cursor.execute(unique_segments_query)

        # 3. Create temporary table with aggregated sessions
        logger.info(f"Creating temporary table {config.android_play_sessions_table_name} with aggregated play sessions...")
        sessions_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_android_play_sessions.sql", params=fetch_params, logger=logger)
        cursor.execute(sessions_query)

        # 4. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.android_stg_play_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={
            **fetch_params,
            "delete_partition_table_name": config.android_stg_play_table_name,
        }, logger=logger)
        cursor.execute(delete_query)

        # 5. Insert new Android play data into the target table
        logger.info(f"Inserting Android play data into {config.android_stg_play_table_name}...")
        insert_android_play_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/stg_android_play.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_android_play_query)
        android_rows = cursor.rowcount

    logger.info(f"Successfully processed partition. Inserted {android_rows} rows")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.android_stg_play_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": android_rows if not is_multi_partition_range else 0,
        },
    )

@dg.asset(
    name="fact_play",
    description="Fact table for clip plays. One row per song session, where each song session is a consecutive sequence of audio events for one user and one clip played.",
    group_name=Group.LISTENING.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=FACT_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(stg_web_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_ios_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_ios_legacy_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_android_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": FACT_PLAY_TABLE_NAME,
        "data_start_date": FACT_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "sla_minutes": 240,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def fact_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: FactPlayConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    params = {
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "web_stg_play_table_name": config.web_stg_play_table_name,
        "ios_stg_play_table_name": config.ios_stg_play_table_name,
        "ios_legacy_stg_play_table_name": config.ios_legacy_stg_play_table_name,
        "android_stg_play_table_name": config.android_stg_play_table_name,
        "fact_play_table_name": config.fact_play_table_name,
    }

    logger.info(f"Processing target partition: {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info("Combining staging tables into fact_play")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.fact_play_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={
            **params,
            "delete_partition_table_name": config.fact_play_table_name,
        }, logger=logger)
        cursor.execute(delete_query)

        # 2. Combine all staging tables
        logger.info("Combining staging tables...")
        combine_query = jinja_formatter.load("src/assets/snowflake/fact/play/queries/combine_staging.sql", params=params, logger=logger)
        cursor.execute(combine_query)
        total_rows = cursor.rowcount

    logger.info(f"Successfully processed partition. Inserted {total_rows} rows from staging tables")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.fact_play_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": total_rows,
        },
    )