from datetime import datetime

import dagster as dg
from dagster import EnvVar

from src.utils.snowflake.constants import Group, PartitionExpr

DIM_CLIP_TABLE_NAME = "DIM_CLIP"
DIM_CLIP_START_DATE = datetime.strptime('2023-05-12', '%Y-%m-%d')

RDS_AUTH_USER_TABLE_NAME = "RDS_AUTH_USER"
RDS_AUTH_USER_START_DATE = datetime.strptime('2023-03-26', '%Y-%m-%d')

USER_SESSION_STATE_TABLE_NAME = "USER_SESSION_STATE"
USER_SESSION_STATE_START_DATE = datetime.strptime('2023-03-26', '%Y-%m-%d')

AGG_USER_DAILY_SPINE_TABLE_NAME = "AGG_USER_DAILY_SPINE"
AGG_USER_DAILY_SPINE_START_DATE = datetime.strptime('2024-06-01', '%Y-%m-%d')

AGG_USER_PLATFORM_DAILY_TABLE_NAME = "AGG_USER_PLATFORM_DAILY"
AGG_USER_PLATFORM_DAILY_START_DATE = datetime.strptime('2024-06-01', '%Y-%m-%d')


user_session_state = dg.AssetSpec(
    key="user_session_state",
    description="User session state table",
    group_name=Group.SESSIONS.value,
    partitions_def=dg.HourlyPartitionsDefinition(start_date=USER_SESSION_STATE_START_DATE),
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": USER_SESSION_STATE_TABLE_NAME,
        "partition_expr": PartitionExpr.HOURLY.value,
    }
)

agg_user_daily_spine = dg.AssetSpec(
    key="agg_user_daily_spine",
    description="Daily user aggregation with calendar spine, providing one row per day per user, even if inactive. Orchestrated via Snowflake task.",
    group_name=Group.USER_METRICS.value,
    partitions_def=dg.DailyPartitionsDefinition(start_date=AGG_USER_DAILY_SPINE_START_DATE, end_offset=0),
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": AGG_USER_DAILY_SPINE_TABLE_NAME,
        "partition_expr": PartitionExpr.DAILY.value,
    }
)

agg_user_platform_daily = dg.AssetSpec(
    key="agg_user_platform_daily",
    description="Active user daily aggregation broken down by platform. Orchestrated via Snowflake task. Source for daily_active_users model.",
    group_name=Group.AGG.value,
    partitions_def=dg.DailyPartitionsDefinition(start_date=AGG_USER_PLATFORM_DAILY_START_DATE, end_offset=0),
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": "prod",  # This table is in prod schema, not SNOWFLAKE_SCHEMA
        "table_name": AGG_USER_PLATFORM_DAILY_TABLE_NAME,
        "partition_expr": PartitionExpr.DAILY.value,
    }
)