WITH pv_status_change AS (
    -- 1. 标准化状态变更值：1表示+1，0表示-1
    SELECT 
        cabinet_id,
        event_time,
        CAST(TO_DATE(event_time) AS STRING) AS op_date,
        CASE WHEN status = 1 THEN 1 ELSE -1 END AS change_val
    FROM grid_status_log
),

pv_occupied_count AS (
    -- 2. 计算每次事件后的累计占用数
    SELECT 
        log.cabinet_id,
        log.op_date,
        log.event_time,
        SUM(log.change_val) OVER (
            PARTITION BY log.cabinet_id 
            ORDER BY log.event_time 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS current_occupied,
        info.total_grids
    FROM pv_status_change log
    JOIN cabinet_info info ON log.cabinet_id = info.cabinet_id
),

pv_state_duration AS (
    -- 3. 使用 LEAD 函数获取当前状态的结束时间，构建时间区间
    SELECT 
        cabinet_id,
        op_date,
        current_occupied,
        total_grids,
        event_time AS start_time,
        -- 如果没有下一条记录，则默认截止到当天最后一秒
        COALESCE(
            LEAD(event_time) OVER (PARTITION BY cabinet_id, op_date ORDER BY event_time),
            CAST(CONCAT(op_date, ' 23:59:59') AS TIMESTAMP)
        ) AS end_time
    FROM pv_occupied_count
),

pv_full_load_minutes AS (
    -- 4. 筛选出“全部占满”的区间，并计算持续分钟数
    SELECT 
        cabinet_id,
        op_date,
        -- 计算差值并转换为分钟数（Spark中TIMESTAMP相减得到秒数，需除以60）
        SUM(CAST(unix_timestamp(end_time) - unix_timestamp(start_time) AS DOUBLE) / 60.0) AS daily_full_duration_minutes
    FROM pv_state_duration
    WHERE current_occupied = total_grids
    GROUP BY cabinet_id, op_date
),

pv_avg_full_load AS (
    -- 5. 计算每个快递柜的日均爆柜时长
    SELECT 
        cabinet_id,
        AVG(daily_full_duration_minutes) AS avg_daily_full_duration_minutes,
        COUNT(DISTINCT op_date) AS active_days
    FROM pv_full_load_minutes
    GROUP BY cabinet_id
)

-- 6. 输出最易爆柜的前 100 个快递柜
SELECT 
    cabinet_id,
    ROUND(avg_daily_full_duration_minutes, 2) AS avg_daily_full_duration_minutes,
    active_days
FROM pv_avg_full_load
ORDER BY avg_daily_full_duration_minutes DESC
LIMIT 100;