from datetime import datetime, timezone, timedelta
import dagster as dg
from dagster import RunRequest
from src.assets.snowflake.personalization.user_top_score_songs.assets import user_top_score_songs_snowflake, USER_TOP_SCORE_SONGS_START_DATE

# Daily job to run user top score songs
user_top_score_songs_daily_job = dg.define_asset_job(
    name="user_top_score_songs_daily_job",
    description="Daily job to calculate and update user top scoring songs based on engagement metrics",
    selection=dg.AssetSelection.assets(user_top_score_songs_snowflake),
    partitions_def=dg.DailyPartitionsDefinition(start_date=USER_TOP_SCORE_SONGS_START_DATE, end_offset=0),
)

# Schedule for the job (daily at 6 AM UTC)
@dg.schedule(
    name="user_top_score_songs_daily_schedule",
    job=user_top_score_songs_daily_job,
    cron_schedule="0 6 * * *",  # Run daily at 6 AM UTC
    default_status=dg.DefaultScheduleStatus.RUNNING if dg.EnvVar("DAGSTER_CLOUD_DEPLOYMENT_NAME").get_value() and not int(dg.EnvVar("DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT").get_value()) == 1 else dg.DefaultScheduleStatus.STOPPED,
)
def user_top_score_songs_daily_schedule(context):
    """Schedule to run user top score songs calculation daily at 6 AM UTC for yesterday's partition"""

    # Always run yesterday's partition in UTC
    yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d")

    return RunRequest(
        partition_key=yesterday,
        tags={
            "schedule_name": "user_top_score_songs_daily_schedule",
            "duration": "12 hours",     # duration for slack alerts
            "dagster/max_runtime": str(3600 * 12),  # 12 hours timeout (in seconds)
        }
    )