INSERT INTO {{ dim_hook_table_name }} (
    ID, PLATFORM, LYRICS_LANGUAGE, CREATED_AT, UPDATED_AT, ORIGINAL_CLIP, STATUS, TITLE,
    START_CLIP_TIMESTAMP, END_CLIP_TIMESTAMP, RENDERED_VIDEO_S3_ID, RENDERED_VIDEO_PREVIEW_S3_ID,
    THUMBNAIL_IMAGE_S3_ID, RAW_VIDEO_UPLOAD_ID, RENDERED_WATERMARKED_VIDEO_S3_ID, USER_ID,
    IS_TEST, ALLOW_COMMENTS, IS_DELETED, CAPTION, SHOW_LYRICS, CONTENT_RATING, CONTENT_RATING_TAGS,
    VIDEO_DURATION, VIDEO_STREAMING_RESOLUTIONS, STAFF_REVIEW_STATUS, IS_AI_GENERATED, IS_UGC,
    HAS_UGC_FACE, HAS_UGC_FACE_IN_FIRST_FRAME, HOOK_CREATION_TYPE_JSON, GEMINI_RATING,
    VIDEO_UPLOAD_IDS, CREATION_SOURCE, METADATA_ID, P_DATE, P_HOUR
)
with a as (
    select
        ID, CREATED_AT, UPDATED_AT, ORIGINAL_CLIP, STATUS, TITLE, START_CLIP_TIMESTAMP, END_CLIP_TIMESTAMP, RENDERED_VIDEO_S3_ID, RENDERED_VIDEO_PREVIEW_S3_ID, THUMBNAIL_IMAGE_S3_ID, RAW_VIDEO_UPLOAD_ID, USER_ID, IS_TEST, ALLOW_COMMENTS, CAPTION, SHOW_LYRICS, CONTENT_RATING, CONTENT_RATING_TAGS, VIDEO_DURATION, VIDEO_STREAMING_RESOLUTIONS, IS_DELETED, RENDERED_WATERMARKED_VIDEO_S3_ID, STAFF_REVIEW_STATUS, VIDEO_UPLOAD_IDS, CREATION_SOURCE, METADATA_ID, P_DATE, P_HOUR
    from rds_video_hook
    where 1=1
        AND IFF(
            '{{ partition_start_date }}' = '{{ partition_end_date }}',
            p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }} AND p_hour < {{ partition_end_hour }},
            (
                (p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }}) OR
                (p_date > '{{ partition_start_date }}' AND p_date < '{{ partition_end_date }}') OR
                (p_date = '{{ partition_end_date }}' AND p_hour < {{ partition_end_hour }})
            )
        )
), b as (
    select
        source as platform,
        properties:hook_id as id
    from bk_nonrealtime_event
    where 1=1
        AND name = 'HookCreateSubmit'
        AND IFF(
            '{{ partition_start_date }}' = '{{ partition_end_date }}',
            p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }} AND p_hour < {{ partition_end_hour }},
            (
                (p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }}) OR
                (p_date > '{{ partition_start_date }}' AND p_date < '{{ partition_end_date }}') OR
                (p_date = '{{ partition_end_date }}' AND p_hour < {{ partition_end_hour }})
            )
        )
), c as (
    select
        id,
        coalesce(json_data:geminiExtractedFeatures:M:hook_content_origin:M:ai_generated:BOOL, json_data:gemini_extracted_features:M:hook_content_origin:M:ai_generated:BOOL) as is_ai_generated,
        coalesce(json_data:geminiExtractedFeatures:M:hook_content_origin:M:ugc:BOOL, json_data:gemini_extracted_features:M:hook_content_origin:M:ugc:BOOL) as is_ugc,
        coalesce(json_data:geminiExtractedFeatures:M:hook_ugc_faces_presence:M:has_ugc_face:BOOL, json_data:gemini_extracted_features:M:hook_ugc_faces_presence:M:has_ugc_face:BOOL) as has_ugc_face,
        coalesce(json_data:geminiExtractedFeatures:M:hook_ugc_faces_presence:M:has_ugc_face_in_first_frame:BOOL, json_data:gemini_extracted_features:M:hook_ugc_faces_presence:M:has_ugc_face_in_first_frame:BOOL) as has_ugc_face_in_first_frame,
        coalesce(json_data:geminiExtractedFeatures:M:hook_creation_type:M:hook_creation_type, json_data:gemini_extracted_features:M:hook_creation_type:M:hook_creation_type) as hook_creation_type_json,
        coalesce(json_data:geminiRating:S, json_data:gemini_rating:S) as gemini_rating
    from ddb_item_info
    where type = 'hook' and  coalesce(json_data:geminiExtractedFeatures, json_data:gemini_extracted_features) is not null
        AND IFF(
            '{{ partition_start_date }}' = '{{ partition_end_date }}',
            p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }} AND p_hour < {{ partition_end_hour }},
            (
                (p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }}) OR
                (p_date > '{{ partition_start_date }}' AND p_date < '{{ partition_end_date }}') OR
                (p_date = '{{ partition_end_date }}' AND p_hour < {{ partition_end_hour }})
            )
        )
), d as (
    select lyrics_language, clip_id
    from dim_clip
    where 1=1
        AND lyrics_language is not null
        AND IFF(
            '{{ partition_start_date }}' = '{{ partition_end_date }}',
            p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }} AND p_hour < {{ partition_end_hour }},
            (
                (p_date = '{{ partition_start_date }}' AND p_hour >= {{ partition_start_hour }}) OR
                (p_date > '{{ partition_start_date }}' AND p_date < '{{ partition_end_date }}') OR
                (p_date = '{{ partition_end_date }}' AND p_hour < {{ partition_end_hour }})
            )
        )
)
select
    a.ID,
    b.platform,
    d.lyrics_language,
    a.CREATED_AT,
    a.UPDATED_AT,
    a.ORIGINAL_CLIP,
    a.STATUS,
    a.TITLE,
    a.START_CLIP_TIMESTAMP,
    a.END_CLIP_TIMESTAMP,
    a.RENDERED_VIDEO_S3_ID,
    a.RENDERED_VIDEO_PREVIEW_S3_ID,
    a.THUMBNAIL_IMAGE_S3_ID,
    a.RAW_VIDEO_UPLOAD_ID,
    a.RENDERED_WATERMARKED_VIDEO_S3_ID,
    a.USER_ID,
    a.IS_TEST,
    a.ALLOW_COMMENTS,
    a.IS_DELETED,
    a.CAPTION,
    a.SHOW_LYRICS,
    a.CONTENT_RATING,
    a.CONTENT_RATING_TAGS,
    a.VIDEO_DURATION,
    a.VIDEO_STREAMING_RESOLUTIONS,
    a.STAFF_REVIEW_STATUS,
    c.is_ai_generated,
    c.is_ugc,
    c.has_ugc_face,
    c.has_ugc_face_in_first_frame,
    c.hook_creation_type_json,
    c.gemini_rating,
    a.VIDEO_UPLOAD_IDS,
    a.CREATION_SOURCE,
    a.METADATA_ID,
    a.P_DATE,
    a.P_HOUR
from a
left join b on a.id = b.id
left join c on a.id = c.id
left join d on a.ORIGINAL_CLIP = d.clip_id;
