WITH base_hourly_stats AS (
    -- Step 1: 统计每个快递柜每天每个小时的投递与取件数量
    SELECT 
        TO_DATE(op_time) AS op_date,
        cabinet_id,
        HOUR(op_time) AS op_hour,
        SUM(CASE WHEN op_type = 'put' THEN 1 ELSE 0 END) AS put_cnt,
        SUM(CASE WHEN op_type = 'get' THEN 1 ELSE 0 END) AS get_cnt
    FROM cabinet_records
    GROUP BY TO_DATE(op_time), cabinet_id, HOUR(op_time)
),

ranked_stats AS (
    -- Step 2: 结合窗口函数，分别对投递和取件数量进行降序排名
    -- 若数量相同，按小时升序排列（取最早的小时）
    SELECT 
        op_date,
        cabinet_id,
        op_hour,
        ROW_NUMBER() OVER(
            PARTITION BY op_date, cabinet_id 
            ORDER BY put_cnt DESC, op_hour ASC
        ) AS put_rank,
        ROW_NUMBER() OVER(
            PARTITION BY op_date, cabinet_id 
            ORDER BY get_cnt DESC, op_hour ASC
        ) AS get_rank
    FROM base_hourly_stats
),

peak_put AS (
    -- Step 3: 筛选出每日投递高峰小时
    SELECT op_date, cabinet_id, op_hour AS peak_put_hour
    FROM ranked_stats
    WHERE put_rank = 1
),

peak_get AS (
    -- Step 4: 筛选出每日取件高峰小时
    SELECT op_date, cabinet_id, op_hour AS peak_get_hour
    FROM ranked_stats
    WHERE get_rank = 1
)

-- Step 5: 关联投递与取件高峰，并计算绝对差值
SELECT 
    p.op_date,
    p.cabinet_id,
    p.peak_put_hour,
    g.peak_get_hour,
    CAST(ABS(p.peak_put_hour - g.peak_get_hour) AS DOUBLE) AS hour_gap
FROM peak_put p
JOIN peak_get g 
  ON p.op_date = g.op_date AND p.cabinet_id = g.cabinet_id
ORDER BY p.op_date, p.cabinet_id;