--*********************
-- 曝光回流收益计算 (B + C 多跳版, BFS 去环, 24h + 48h)
-- 同时输出 24h 和 48h 窗口的 B/C/mids
--*********************

WITH
--========================================
-- 1. 基础数据准备
--========================================

-- 1.1 回流数据 24h
t_return AS (
    SELECT  *
            ,CONCAT(dthh,":",shareid,":",vid,":",dthh_id) AS id
    FROM    (
                SELECT  CONCAT(year,month,day,hour) AS dthh
                        ,apptype
                        ,machinecode AS mid
                        ,clickobjectid AS vid
                        ,sessionid
                        ,subsessionid
                        ,shareid
                        ,rootshareid
                        ,CAST(clienttimestamp / 1000 AS BIGINT) AS ts
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),apptype,machinecode,clickobjectid,sessionid,subsessionid,shareid,rootshareid ORDER BY clienttimestamp DESC ) AS rn
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),shareid,clickobjectid ORDER BY clienttimestamp ) AS dthh_id
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) BETWEEN '${dt}${hh}' AND TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) + 3600 * 24),'YYYYMMDDHH')
                AND     __topic__ = 'click'
                AND     apptype IS NOT NULL
                AND     apptype NOT IN ('12')
                AND     machinecode IS NOT NULL
                AND     clickobjectid IS NOT NULL
                AND     pagesource REGEXP "-pages/user-videos-share$"
            ) t
    WHERE   rn = 1
)

-- 1.1b 回流数据 48h
,t_return_48h AS (
    SELECT  *
            ,CONCAT(dthh,":",shareid,":",vid,":",dthh_id) AS id
    FROM    (
                SELECT  CONCAT(year,month,day,hour) AS dthh
                        ,apptype
                        ,machinecode AS mid
                        ,clickobjectid AS vid
                        ,sessionid
                        ,subsessionid
                        ,shareid
                        ,rootshareid
                        ,CAST(clienttimestamp / 1000 AS BIGINT) AS ts
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),apptype,machinecode,clickobjectid,sessionid,subsessionid,shareid,rootshareid ORDER BY clienttimestamp DESC ) AS rn
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),shareid,clickobjectid ORDER BY clienttimestamp ) AS dthh_id
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) BETWEEN '${dt}${hh}' AND TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) + 3600 * 48),'YYYYMMDDHH')
                AND     __topic__ = 'click'
                AND     apptype IS NOT NULL
                AND     apptype NOT IN ('12')
                AND     machinecode IS NOT NULL
                AND     clickobjectid IS NOT NULL
                AND     pagesource REGEXP "-pages/user-videos-share$"
            ) t
    WHERE   rn = 1
)

-- 1.2 分享数据 24h
,t_share_from_sharelog AS (
    SELECT  *
    FROM    (
                SELECT  CONCAT(year,month,day,hour) AS dthh
                        ,apptype
                        ,machinecode AS mid
                        ,shareobjectid AS vid
                        ,sessionid
                        ,subsessionid
                        ,pagesource
                        ,shareid
                        ,CAST(clienttimestamp / 1000 AS BIGINT) AS ts
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),apptype,machinecode,shareobjectid,sessionid,subsessionid,pagesource,shareid ORDER BY clienttimestamp DESC ) AS rn
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) BETWEEN '${dt}${hh}' AND TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) + 3600 * 24),'YYYYMMDDHH')
                AND     __topic__ = 'share'
                AND     apptype IS NOT NULL
                AND     apptype NOT IN ('12')
                AND     machinecode IS NOT NULL
                AND     shareobjectid IS NOT NULL
            ) t
    WHERE   rn = 1
)

-- 1.2b 分享数据 48h
,t_share_from_sharelog_48h AS (
    SELECT  *
    FROM    (
                SELECT  CONCAT(year,month,day,hour) AS dthh
                        ,apptype
                        ,machinecode AS mid
                        ,shareobjectid AS vid
                        ,sessionid
                        ,subsessionid
                        ,pagesource
                        ,shareid
                        ,CAST(clienttimestamp / 1000 AS BIGINT) AS ts
                        ,ROW_NUMBER() OVER (PARTITION BY CONCAT(year,month,day,hour),apptype,machinecode,shareobjectid,sessionid,subsessionid,pagesource,shareid ORDER BY clienttimestamp DESC ) AS rn
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) BETWEEN '${dt}${hh}' AND TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) + 3600 * 48),'YYYYMMDDHH')
                AND     __topic__ = 'share'
                AND     apptype IS NOT NULL
                AND     apptype NOT IN ('12')
                AND     machinecode IS NOT NULL
                AND     shareobjectid IS NOT NULL
            ) t
    WHERE   rn = 1
)

-- 1.3 曝光数据（共用）
,t_exposure AS (
    SELECT  dthh_id
            ,dthh
            ,apptype
            ,uid
            ,mid
            ,vid
            ,sessionid
            ,subsessionid
            ,pagesource
            ,ts
            ,id
            ,dt
            ,hh
    FROM    loghubods.dwd_recsys_alg_exposure_base_view_20250402
    WHERE   CONCAT(dt,hh) = '${dt}${hh}'
)

-- 1.4 详情页曝光（共用）
,t_exposure_detail AS (
    SELECT  *
    FROM    t_exposure
    WHERE   pagesource REGEXP "-pages/user-videos-detail$|pages/detail-recommend$"
)

--========================================
-- 2. 曝光关联分享 24h
--========================================
,t_normal_share_exposure AS (
    SELECT  *
    FROM    (
        SELECT  s.dthh
                ,s.apptype
                ,s.mid
                ,s.vid
                ,s.sessionid
                ,s.subsessionid
                ,s.pagesource
                ,s.shareid
                ,s.ts
                ,e.id AS exposure_id
                ,e.ts AS exposure_ts
                ,CASE
                    WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 1
                    WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 2
                    WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource THEN 3
                    WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource THEN 4
                    WHEN s.subsessionid = e.subsessionid THEN 5
                    WHEN s.sessionid = e.sessionid THEN 6
                END AS match_priority
                ,ROW_NUMBER() OVER (
                    PARTITION BY s.dthh,s.apptype,s.mid,s.vid,s.sessionid,s.subsessionid,s.pagesource,s.shareid
                    ORDER BY
                        CASE
                            WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 1
                            WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 2
                            WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource THEN 3
                            WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource THEN 4
                            WHEN s.subsessionid = e.subsessionid THEN 5
                            WHEN s.sessionid = e.sessionid THEN 6
                        END
                        ,e.ts DESC
                ) AS rn
        FROM    t_share_from_sharelog s
        LEFT JOIN t_exposure e
        ON      s.apptype = e.apptype
        AND     s.mid = e.mid
        AND     s.vid = e.vid
        AND     (s.subsessionid = e.subsessionid OR s.sessionid = e.sessionid)
        WHERE   s.pagesource NOT REGEXP "pages/detail-user-videos-share-recommend$"
    ) t
    WHERE   rn = 1
)

,t_no_normal_share_exposure AS (
    SELECT  *
    FROM    (
        SELECT  s.dthh
                ,s.apptype
                ,s.mid
                ,s.vid
                ,s.sessionid
                ,s.subsessionid
                ,s.pagesource
                ,s.shareid
                ,s.ts
                ,e.id AS exposure_id
                ,e.ts AS exposure_ts
                ,CASE
                    WHEN s.subsessionid = e.subsessionid AND s.ts >= e.ts THEN 1
                    WHEN s.sessionid = e.sessionid AND s.ts >= e.ts THEN 2
                    WHEN s.subsessionid = e.subsessionid THEN 3
                    WHEN s.sessionid = e.sessionid THEN 4
                END AS match_priority
                ,ROW_NUMBER() OVER (
                    PARTITION BY s.dthh,s.apptype,s.mid,s.vid,s.sessionid,s.subsessionid,s.pagesource,s.shareid
                    ORDER BY
                        CASE
                            WHEN s.subsessionid = e.subsessionid AND s.ts >= e.ts THEN 1
                            WHEN s.sessionid = e.sessionid AND s.ts >= e.ts THEN 2
                            WHEN s.subsessionid = e.subsessionid THEN 3
                            WHEN s.sessionid = e.sessionid THEN 4
                        END
                        ,e.ts DESC
                ) AS rn
        FROM    t_share_from_sharelog s
        LEFT JOIN t_exposure_detail e
        ON      s.apptype = e.apptype
        AND     s.mid = e.mid
        AND     s.vid = e.vid
        AND     (s.subsessionid = e.subsessionid OR s.sessionid = e.sessionid)
        WHERE   s.pagesource REGEXP "pages/detail-user-videos-share-recommend$"
    ) t
    WHERE   rn = 1
)

,t_share_exposure AS (
    SELECT  dthh, apptype, mid, vid, sessionid, subsessionid, pagesource, shareid, ts, exposure_id, exposure_ts
    FROM    t_normal_share_exposure
    UNION ALL
    SELECT  dthh, apptype, mid, vid, sessionid, subsessionid, pagesource, shareid, ts, exposure_id, exposure_ts
    FROM    t_no_normal_share_exposure
)

--========================================
-- 2b. 曝光关联分享 48h
--========================================
,t_normal_share_exposure_48h AS (
    SELECT  *
    FROM    (
        SELECT  s.dthh
                ,s.apptype
                ,s.mid
                ,s.vid
                ,s.sessionid
                ,s.subsessionid
                ,s.pagesource
                ,s.shareid
                ,s.ts
                ,e.id AS exposure_id
                ,e.ts AS exposure_ts
                ,CASE
                    WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 1
                    WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 2
                    WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource THEN 3
                    WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource THEN 4
                    WHEN s.subsessionid = e.subsessionid THEN 5
                    WHEN s.sessionid = e.sessionid THEN 6
                END AS match_priority
                ,ROW_NUMBER() OVER (
                    PARTITION BY s.dthh,s.apptype,s.mid,s.vid,s.sessionid,s.subsessionid,s.pagesource,s.shareid
                    ORDER BY
                        CASE
                            WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 1
                            WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource AND s.ts >= e.ts THEN 2
                            WHEN s.subsessionid = e.subsessionid AND s.pagesource = e.pagesource THEN 3
                            WHEN s.sessionid = e.sessionid AND s.pagesource = e.pagesource THEN 4
                            WHEN s.subsessionid = e.subsessionid THEN 5
                            WHEN s.sessionid = e.sessionid THEN 6
                        END
                        ,e.ts DESC
                ) AS rn
        FROM    t_share_from_sharelog_48h s
        LEFT JOIN t_exposure e
        ON      s.apptype = e.apptype
        AND     s.mid = e.mid
        AND     s.vid = e.vid
        AND     (s.subsessionid = e.subsessionid OR s.sessionid = e.sessionid)
        WHERE   s.pagesource NOT REGEXP "pages/detail-user-videos-share-recommend$"
    ) t
    WHERE   rn = 1
)

,t_no_normal_share_exposure_48h AS (
    SELECT  *
    FROM    (
        SELECT  s.dthh
                ,s.apptype
                ,s.mid
                ,s.vid
                ,s.sessionid
                ,s.subsessionid
                ,s.pagesource
                ,s.shareid
                ,s.ts
                ,e.id AS exposure_id
                ,e.ts AS exposure_ts
                ,CASE
                    WHEN s.subsessionid = e.subsessionid AND s.ts >= e.ts THEN 1
                    WHEN s.sessionid = e.sessionid AND s.ts >= e.ts THEN 2
                    WHEN s.subsessionid = e.subsessionid THEN 3
                    WHEN s.sessionid = e.sessionid THEN 4
                END AS match_priority
                ,ROW_NUMBER() OVER (
                    PARTITION BY s.dthh,s.apptype,s.mid,s.vid,s.sessionid,s.subsessionid,s.pagesource,s.shareid
                    ORDER BY
                        CASE
                            WHEN s.subsessionid = e.subsessionid AND s.ts >= e.ts THEN 1
                            WHEN s.sessionid = e.sessionid AND s.ts >= e.ts THEN 2
                            WHEN s.subsessionid = e.subsessionid THEN 3
                            WHEN s.sessionid = e.sessionid THEN 4
                        END
                        ,e.ts DESC
                ) AS rn
        FROM    t_share_from_sharelog_48h s
        LEFT JOIN t_exposure_detail e
        ON      s.apptype = e.apptype
        AND     s.mid = e.mid
        AND     s.vid = e.vid
        AND     (s.subsessionid = e.subsessionid OR s.sessionid = e.sessionid)
        WHERE   s.pagesource REGEXP "pages/detail-user-videos-share-recommend$"
    ) t
    WHERE   rn = 1
)

,t_share_exposure_48h AS (
    SELECT  dthh, apptype, mid, vid, sessionid, subsessionid, pagesource, shareid, ts, exposure_id, exposure_ts
    FROM    t_normal_share_exposure_48h
    UNION ALL
    SELECT  dthh, apptype, mid, vid, sessionid, subsessionid, pagesource, shareid, ts, exposure_id, exposure_ts
    FROM    t_no_normal_share_exposure_48h
)

--========================================
-- 3. 分享关联回流 + B 值计算 (24h)
--========================================
,t_share_return AS (
    SELECT  se.exposure_id
            ,se.shareid
            ,se.vid
            ,se.apptype
            ,se.subsessionid
            ,r.subsessionid AS return_subsessionid
            ,r.mid AS return_mid
    FROM    t_share_exposure se
    JOIN    t_return r
    ON      se.shareid = r.rootshareid
    AND     se.vid = r.vid
    AND     se.apptype = r.apptype
)

,t_exposure_bn AS (
    SELECT  e.id AS exposure_id
            ,e.subsessionid
            ,e.ts
            ,e.vid
            ,e.uid
            ,e.mid
            ,COALESCE(bn.B, 0) AS B
            ,bn.B_mids
    FROM    t_exposure e
    LEFT JOIN (
        SELECT  exposure_id
                ,COUNT(DISTINCT return_mid) AS B
                ,COLLECT_SET(return_mid) AS B_mids
        FROM    t_share_return
        GROUP BY exposure_id
    ) bn
    ON      e.id = bn.exposure_id
)

--========================================
-- 3b. 分享关联回流 + B 值计算 (48h)
--========================================
,t_share_return_48h AS (
    SELECT  se.exposure_id
            ,se.shareid
            ,se.vid
            ,se.apptype
            ,se.subsessionid
            ,r.subsessionid AS return_subsessionid
            ,r.mid AS return_mid
    FROM    t_share_exposure_48h se
    JOIN    t_return_48h r
    ON      se.shareid = r.rootshareid
    AND     se.vid = r.vid
    AND     se.apptype = r.apptype
)

,t_exposure_bn_48h AS (
    SELECT  e.id AS exposure_id
            ,e.subsessionid
            ,COALESCE(bn.B, 0) AS B
            ,bn.B_mids
    FROM    t_exposure e
    LEFT JOIN (
        SELECT  exposure_id
                ,COUNT(DISTINCT return_mid) AS B
                ,COLLECT_SET(return_mid) AS B_mids
        FROM    t_share_return_48h
        GROUP BY exposure_id
    ) bn
    ON      e.id = bn.exposure_id
)

--========================================
-- 4. 多跳计算 24h（BFS frontier + subsession 去环）
--   MID 跨层去重在最终输出用 ARRAY_EXCEPT 计算
--========================================

,t_frontier_1 AS (
    SELECT DISTINCT exposure_id AS source_id, return_subsessionid AS reached_sub
    FROM   t_share_return
)
,t_c1 AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_1
    FROM    t_frontier_1 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c1_mids AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_1_mids
    FROM    t_frontier_1 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

,t_frontier_2 AS (
    SELECT DISTINCT f1.source_id, sr2.return_subsessionid AS reached_sub
    FROM    t_frontier_1 f1
    JOIN    t_exposure_bn eb1 ON f1.reached_sub = eb1.subsessionid
    JOIN    t_share_return sr2 ON eb1.exposure_id = sr2.exposure_id
    LEFT JOIN t_frontier_1 v1
        ON  f1.source_id = v1.source_id
        AND sr2.return_subsessionid = v1.reached_sub
    WHERE   v1.source_id IS NULL
)
,t_c2 AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_2
    FROM    t_frontier_2 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c2_mids AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_2_mids
    FROM    t_frontier_2 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

,t_frontier_3 AS (
    SELECT DISTINCT f2.source_id, sr3.return_subsessionid AS reached_sub
    FROM    t_frontier_2 f2
    JOIN    t_exposure_bn eb2 ON f2.reached_sub = eb2.subsessionid
    JOIN    t_share_return sr3 ON eb2.exposure_id = sr3.exposure_id
    LEFT JOIN t_frontier_1 v1
        ON  f2.source_id = v1.source_id
        AND sr3.return_subsessionid = v1.reached_sub
    LEFT JOIN t_frontier_2 v2
        ON  f2.source_id = v2.source_id
        AND sr3.return_subsessionid = v2.reached_sub
    WHERE   v1.source_id IS NULL AND v2.source_id IS NULL
)
,t_c3 AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_3
    FROM    t_frontier_3 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c3_mids AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_3_mids
    FROM    t_frontier_3 f
    JOIN    t_exposure_bn eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

--========================================
-- 4b. 多跳计算 48h（BFS frontier + subsession 去环）
--========================================

,t_frontier_1_48h AS (
    SELECT DISTINCT exposure_id AS source_id, return_subsessionid AS reached_sub
    FROM   t_share_return_48h
)
,t_c1_48h AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_1
    FROM    t_frontier_1_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c1_mids_48h AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_1_mids
    FROM    t_frontier_1_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return_48h sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

,t_frontier_2_48h AS (
    SELECT DISTINCT f1.source_id, sr2.return_subsessionid AS reached_sub
    FROM    t_frontier_1_48h f1
    JOIN    t_exposure_bn_48h eb1 ON f1.reached_sub = eb1.subsessionid
    JOIN    t_share_return_48h sr2 ON eb1.exposure_id = sr2.exposure_id
    LEFT JOIN t_frontier_1_48h v1
        ON  f1.source_id = v1.source_id
        AND sr2.return_subsessionid = v1.reached_sub
    WHERE   v1.source_id IS NULL
)
,t_c2_48h AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_2
    FROM    t_frontier_2_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c2_mids_48h AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_2_mids
    FROM    t_frontier_2_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return_48h sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

,t_frontier_3_48h AS (
    SELECT DISTINCT f2.source_id, sr3.return_subsessionid AS reached_sub
    FROM    t_frontier_2_48h f2
    JOIN    t_exposure_bn_48h eb2 ON f2.reached_sub = eb2.subsessionid
    JOIN    t_share_return_48h sr3 ON eb2.exposure_id = sr3.exposure_id
    LEFT JOIN t_frontier_1_48h v1
        ON  f2.source_id = v1.source_id
        AND sr3.return_subsessionid = v1.reached_sub
    LEFT JOIN t_frontier_2_48h v2
        ON  f2.source_id = v2.source_id
        AND sr3.return_subsessionid = v2.reached_sub
    WHERE   v1.source_id IS NULL AND v2.source_id IS NULL
)
,t_c3_48h AS (
    SELECT  f.source_id AS exposure_id, SUM(eb.B) AS C_3
    FROM    t_frontier_3_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    GROUP BY f.source_id
)
,t_c3_mids_48h AS (
    SELECT  f.source_id AS exposure_id
            ,COLLECT_SET(sr.return_mid) AS C_3_mids
    FROM    t_frontier_3_48h f
    JOIN    t_exposure_bn_48h eb ON f.reached_sub = eb.subsessionid
    JOIN    t_share_return_48h sr ON eb.exposure_id = sr.exposure_id
    GROUP BY f.source_id
)

--========================================
-- 5. 最终输出
--========================================
SELECT  e.id AS exposure_id
        ,e.vid
        ,v.title AS video_title
        ,e.uid
        ,e.mid
        ,e.ts
        -- 24h 非去重
        ,CASE WHEN se.exposure_id IS NOT NULL THEN 1 ELSE 0 END AS is_share
        ,COALESCE(bn.B, 0) AS B
        ,CONCAT_WS(',', bn.B_mids) AS B_mids
        ,COALESCE(c1.C_1, 0) AS C_1
        ,CONCAT_WS(',', c1m.C_1_mids) AS C_1_mids
        ,COALESCE(c2.C_2, 0) AS C_2
        ,CONCAT_WS(',', c2m.C_2_mids) AS C_2_mids
        ,COALESCE(c3.C_3, 0) AS C_3
        ,CONCAT_WS(',', c3m.C_3_mids) AS C_3_mids
        ,COALESCE(bn.B, 0) + COALESCE(c1.C_1, 0) + COALESCE(c2.C_2, 0) + COALESCE(c3.C_3, 0) AS V_total
        -- 24h 去重（ARRAY_EXCEPT 跨层排除已计入的 MID）
        ,SIZE(c1m.C_1_mids) AS C_1_distinct
        ,IF(c1m.C_1_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(c1m.C_1_mids, COALESCE(bn.B_mids, ARRAY('')))), 0) AS C_1_dedup
        ,IF(c2m.C_2_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(c2m.C_2_mids, COALESCE(bn.B_mids, ARRAY(''))), COALESCE(c1m.C_1_mids, ARRAY('')))), 0) AS C_2_dedup
        ,IF(c3m.C_3_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(ARRAY_EXCEPT(c3m.C_3_mids, COALESCE(bn.B_mids, ARRAY(''))), COALESCE(c1m.C_1_mids, ARRAY(''))), COALESCE(c2m.C_2_mids, ARRAY('')))), 0) AS C_3_dedup
        ,COALESCE(bn.B, 0)
            + IF(c1m.C_1_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(c1m.C_1_mids, COALESCE(bn.B_mids, ARRAY('')))), 0)
            + IF(c2m.C_2_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(c2m.C_2_mids, COALESCE(bn.B_mids, ARRAY(''))), COALESCE(c1m.C_1_mids, ARRAY('')))), 0)
            + IF(c3m.C_3_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(ARRAY_EXCEPT(c3m.C_3_mids, COALESCE(bn.B_mids, ARRAY(''))), COALESCE(c1m.C_1_mids, ARRAY(''))), COALESCE(c2m.C_2_mids, ARRAY('')))), 0)
            AS V_total_dedup
        -- 48h 非去重
        ,CASE WHEN se48.exposure_id IS NOT NULL THEN 1 ELSE 0 END AS is_share_48h
        ,COALESCE(bn48.B, 0) AS B_48h
        ,CONCAT_WS(',', bn48.B_mids) AS B_mids_48h
        ,COALESCE(c1_48.C_1, 0) AS C_1_48h
        ,CONCAT_WS(',', c1m48.C_1_mids) AS C_1_mids_48h
        ,COALESCE(c2_48.C_2, 0) AS C_2_48h
        ,CONCAT_WS(',', c2m48.C_2_mids) AS C_2_mids_48h
        ,COALESCE(c3_48.C_3, 0) AS C_3_48h
        ,CONCAT_WS(',', c3m48.C_3_mids) AS C_3_mids_48h
        ,COALESCE(bn48.B, 0) + COALESCE(c1_48.C_1, 0) + COALESCE(c2_48.C_2, 0) + COALESCE(c3_48.C_3, 0) AS V_total_48h
        -- 48h 去重
        ,IF(c1m48.C_1_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(c1m48.C_1_mids, COALESCE(bn48.B_mids, ARRAY('')))), 0) AS C_1_dedup_48h
        ,IF(c2m48.C_2_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(c2m48.C_2_mids, COALESCE(bn48.B_mids, ARRAY(''))), COALESCE(c1m48.C_1_mids, ARRAY('')))), 0) AS C_2_dedup_48h
        ,IF(c3m48.C_3_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(ARRAY_EXCEPT(c3m48.C_3_mids, COALESCE(bn48.B_mids, ARRAY(''))), COALESCE(c1m48.C_1_mids, ARRAY(''))), COALESCE(c2m48.C_2_mids, ARRAY('')))), 0) AS C_3_dedup_48h
        ,COALESCE(bn48.B, 0)
            + IF(c1m48.C_1_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(c1m48.C_1_mids, COALESCE(bn48.B_mids, ARRAY('')))), 0)
            + IF(c2m48.C_2_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(c2m48.C_2_mids, COALESCE(bn48.B_mids, ARRAY(''))), COALESCE(c1m48.C_1_mids, ARRAY('')))), 0)
            + IF(c3m48.C_3_mids IS NOT NULL, SIZE(ARRAY_EXCEPT(ARRAY_EXCEPT(ARRAY_EXCEPT(c3m48.C_3_mids, COALESCE(bn48.B_mids, ARRAY(''))), COALESCE(c1m48.C_1_mids, ARRAY(''))), COALESCE(c2m48.C_2_mids, ARRAY('')))), 0)
            AS V_total_dedup_48h
FROM    t_exposure e
-- 24h joins
LEFT JOIN (SELECT DISTINCT exposure_id FROM t_share_exposure) se
ON      e.id = se.exposure_id
LEFT JOIN t_exposure_bn bn
ON      e.id = bn.exposure_id
LEFT JOIN t_c1 c1
ON      e.id = c1.exposure_id
LEFT JOIN t_c1_mids c1m
ON      e.id = c1m.exposure_id
LEFT JOIN t_c2 c2
ON      e.id = c2.exposure_id
LEFT JOIN t_c2_mids c2m
ON      e.id = c2m.exposure_id
LEFT JOIN t_c3 c3
ON      e.id = c3.exposure_id
LEFT JOIN t_c3_mids c3m
ON      e.id = c3m.exposure_id
-- 48h joins
LEFT JOIN (SELECT DISTINCT exposure_id FROM t_share_exposure_48h) se48
ON      e.id = se48.exposure_id
LEFT JOIN t_exposure_bn_48h bn48
ON      e.id = bn48.exposure_id
LEFT JOIN t_c1_48h c1_48
ON      e.id = c1_48.exposure_id
LEFT JOIN t_c1_mids_48h c1m48
ON      e.id = c1m48.exposure_id
LEFT JOIN t_c2_48h c2_48
ON      e.id = c2_48.exposure_id
LEFT JOIN t_c2_mids_48h c2m48
ON      e.id = c2m48.exposure_id
LEFT JOIN t_c3_48h c3_48
ON      e.id = c3_48.exposure_id
LEFT JOIN t_c3_mids_48h c3m48
ON      e.id = c3m48.exposure_id
-- video info
LEFT JOIN videoods.wx_video v
ON      e.vid = CAST(v.id AS STRING)
;
