--*********************
-- 曝光回流收益计算 (B + C 多跳版, BFS 去环)
-- 数学公式:
--   B_i      = 不换视频回流人数（用 rootshareid 计算）
--   C_1(i)   = Σ B_j, j ∈ frontier_1（直接触达的 subsession）
--   C_2(i)   = Σ B_k, k ∈ frontier_2（排除 hop1 已访问的）
--   C_3(i)   = 同理（排除 hop1 + hop2 已访问的）
--   V_total  = B + C_1 + C_2 + C_3
--
-- 改进: BFS frontier + anti-join 去环
--   每个 subsession 只在最短跳数被计入，不重复
--   解决 C_1=C_2=C_3 的环路膨胀问题
--*********************

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

-- 1.1 回流数据（用户通过分享链接回流）
-- 时间范围：${dt}${hh} 往后 24 小时
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.2 分享数据
-- 时间范围：${dt}${hh} 往后 24 小时
,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.3 曝光数据
-- 时间范围：${dt}${hh} 单小时（曝光起点）
,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. 曝光关联分享（单次 JOIN + 优先级选择）
--========================================

-- 2.1 常规分享关联曝光（一次性匹配所有优先级）
,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
                -- 优先级：subsession+pagesource+ts > session+pagesource+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
)

-- 2.2 非常规分享关联曝光（detail页面，一次性匹配）
,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
)

--========================================
-- 3. 合并所有分享-曝光关联
--========================================
,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
)

--========================================
-- 4. 分享关联回流 + B 值计算
--========================================

-- 4.1 分享关联回流（exposure_id → return_subsessionid）
,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
)

-- 4.2 每个曝光的 B 值和 bn_subsessions
,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.bn_subsessions
    FROM    t_exposure e
    LEFT JOIN (
        SELECT  exposure_id
                ,COUNT(DISTINCT return_mid) AS B
                ,COLLECT_SET(return_subsessionid) AS bn_subsessions
        FROM    t_share_return
        GROUP BY exposure_id
    ) bn
    ON      e.id = bn.exposure_id
)

--========================================
-- 5. 多跳计算（BFS frontier + anti-join 去环）
--   每个 subsession 只在最短跳数被计入
--========================================

-- Frontier 1: 源曝光直接触达的 subsession（去重）
,t_frontier_1 AS (
    SELECT DISTINCT exposure_id AS source_id, return_subsessionid AS reached_sub
    FROM   t_share_return
)

-- C_1: hop1 触达的 subsession 中曝光的 B 之和
,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
)

-- C_1 涉及的回流 MID（去重）
,t_c1_mids AS (
    SELECT  f.source_id AS exposure_id
            ,CONCAT_WS(',', 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
)

-- Frontier 2: 从 hop1 延伸一跳，排除 hop1 已访问的 subsession
,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
)

-- C_2: hop2 触达的 subsession 中曝光的 B 之和
,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
)

-- C_2 涉及的回流 MID（去重）
,t_c2_mids AS (
    SELECT  f.source_id AS exposure_id
            ,CONCAT_WS(',', 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
)

-- Frontier 3: 从 hop2 延伸一跳，排除 hop1 + hop2 已访问的
,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
)

-- C_3: hop3 触达的 subsession 中曝光的 B 之和
,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
)

-- C_3 涉及的回流 MID（去重）
,t_c3_mids AS (
    SELECT  f.source_id AS exposure_id
            ,CONCAT_WS(',', 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
)

--========================================
-- 6. 最终输出
--========================================
SELECT  e.id AS exposure_id
        ,e.vid
        ,v.title AS video_title
        ,e.uid
        ,e.mid
        ,e.ts
        ,CASE WHEN se.exposure_id IS NOT NULL THEN 1 ELSE 0 END AS is_share
        ,COALESCE(bn.B, 0) AS B
        ,COALESCE(c1.C_1, 0) AS C_1
        ,c1m.C_1_mids
        ,COALESCE(c2.C_2, 0) AS C_2
        ,c2m.C_2_mids
        ,COALESCE(c3.C_3, 0) AS C_3
        ,c3m.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
FROM    t_exposure e
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
LEFT JOIN videoods.wx_video v
ON      e.vid = CAST(v.id AS STRING)
;
