-- ${bizdate} DAU安全行为累计分五档及当日分享/回流贡献
-- 得分窗口：${history_start_dt}-${history_end_dt}，不包含统计日
-- S = 有效播放次数 + ${active_weight} * 活跃天数 + 分享次数
--     + 50 * I(窗口内出现过1067或1095来源场景)
-- 分位点按${bizdate} DAU用户人数从低分到高分计算，同分用户保持同档。

WITH dau AS (
    SELECT DISTINCT machinecode AS mid
    FROM loghubods.useractive_log
    WHERE dt = '${bizdate}'
      AND businesstype = 'path'
      AND machinecode IS NOT NULL
      AND machinecode <> ''
),
history_active AS (
    SELECT
        a.machinecode AS mid,
        COUNT(DISTINCT a.dt) AS active_day_cnt
    FROM loghubods.useractive_log a
    JOIN dau d
      ON a.machinecode = d.mid
    WHERE a.dt BETWEEN '${history_start_dt}' AND '${history_end_dt}'
      AND a.businesstype = 'path'
    GROUP BY a.machinecode
),
history_real_play AS (
    SELECT
        p.mid,
        COUNT(*) AS real_play_cnt
    FROM loghubods.video_play_log p
    JOIN dau d
      ON p.mid = d.mid
    WHERE p.dt BETWEEN '${history_start_dt}' AND '${history_end_dt}'
      AND p.businesstype = 'videoRealPlay'
    GROUP BY p.mid
),
history_share AS (
    SELECT
        v.mid,
        COUNT(*) AS share_cnt
    FROM loghubods.video_action_log_applet v
    JOIN dau d
      ON v.mid = d.mid
    WHERE v.dt BETWEEN '${history_start_dt}' AND '${history_end_dt}'
      AND v.business = 'videoShareFriend'
      AND v.businesstype = 'videoShareFriend'
    GROUP BY v.mid
),
history_paid_source AS (
    SELECT
        s.machinecode AS mid,
        1 AS has_paid_source
    FROM loghubods.user_share_log s
    JOIN dau d
      ON s.machinecode = d.mid
    WHERE s.dt BETWEEN '${history_start_dt}' AND '${history_end_dt}'
      AND s.topic = 'click'
      AND s.hotsencetype IN ('1067', '1095')
    GROUP BY s.machinecode
),
user_score AS (
    SELECT
        d.mid,
        COALESCE(p.real_play_cnt, 0) AS real_play_cnt,
        COALESCE(a.active_day_cnt, 0) AS active_day_cnt,
        COALESCE(s.share_cnt, 0) AS history_share_cnt,
        COALESCE(x.has_paid_source, 0) AS has_paid_source,
        COALESCE(p.real_play_cnt, 0)
          + ${active_weight} * COALESCE(a.active_day_cnt, 0)
          + COALESCE(s.share_cnt, 0)
          + CASE WHEN COALESCE(x.has_paid_source, 0) = 1 THEN 50 ELSE 0 END
          AS safety_score
    FROM dau d
    LEFT JOIN history_active a
      ON d.mid = a.mid
    LEFT JOIN history_real_play p
      ON d.mid = p.mid
    LEFT JOIN history_share s
      ON d.mid = s.mid
    LEFT JOIN history_paid_source x
      ON d.mid = x.mid
),
score_counts AS (
    SELECT safety_score, COUNT(*) AS score_uv
    FROM user_score
    GROUP BY safety_score
),
score_cumulative AS (
    SELECT
        safety_score,
        score_uv,
        SUM(score_uv) OVER (
            ORDER BY safety_score
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS cumulative_uv,
        SUM(score_uv) OVER () AS total_uv
    FROM score_counts
),
thresholds AS (
    SELECT
        MIN(CASE WHEN cumulative_uv >= total_uv * 0.20 THEN safety_score END) AS p20_score,
        MIN(CASE WHEN cumulative_uv >= total_uv * 0.40 THEN safety_score END) AS p40_score,
        MIN(CASE WHEN cumulative_uv >= total_uv * 0.60 THEN safety_score END) AS p60_score,
        MIN(CASE WHEN cumulative_uv >= total_uv * 0.80 THEN safety_score END) AS p80_score,
        MAX(total_uv) AS total_dau
    FROM score_cumulative
),
scored_dau AS (
    SELECT
        u.*,
        CASE
            WHEN u.safety_score > t.p80_score THEN 1
            WHEN u.safety_score > t.p60_score THEN 2
            WHEN u.safety_score > t.p40_score THEN 3
            WHEN u.safety_score > t.p20_score THEN 4
            ELSE 5
        END AS band_order
    FROM user_score u
    CROSS JOIN thresholds t
),
score_bands AS (
    SELECT 1 AS band_order, '安全得分头部20%' AS quantile_band
    UNION ALL
    SELECT 2 AS band_order, '安全得分20%-40%' AS quantile_band
    UNION ALL
    SELECT 3 AS band_order, '安全得分40%-60%' AS quantile_band
    UNION ALL
    SELECT 4 AS band_order, '安全得分60%-80%' AS quantile_band
    UNION ALL
    SELECT 5 AS band_order, '安全得分尾部20%' AS quantile_band
),
daily_share_by_mid AS (
    SELECT
        v.mid,
        COUNT(*) AS share_cnt
    FROM loghubods.video_action_log_applet v
    JOIN dau d
      ON v.mid = d.mid
    WHERE v.dt = '${bizdate}'
      AND v.business = 'videoShareFriend'
      AND v.businesstype = 'videoShareFriend'
    GROUP BY v.mid
),
daily_ad_exposure_mids AS (
    SELECT DISTINCT
        a.machinecode AS mid
    FROM loghubods.ad_action_log_own a
    JOIN dau d
      ON a.machinecode = d.mid
    WHERE a.dt = '${bizdate}'
      AND a.businesstype = 'adView'
      AND a.ownAdSystemType = 'ownPlatform'
),
daily_ad_click_events AS (
    SELECT DISTINCT
        a.machinecode AS mid,
        a.pqtid
    FROM loghubods.ad_action_log_own a
    JOIN dau d
      ON a.machinecode = d.mid
    WHERE a.dt = '${bizdate}'
      AND a.businesstype = 'adClick'
      AND a.ownAdSystemType = 'ownPlatform'
),
daily_ad_click_mids AS (
    SELECT DISTINCT mid
    FROM daily_ad_click_events
),
daily_conversion_pqtids AS (
    SELECT DISTINCT pqtid
    FROM loghubods.ad_own_open_conv
    WHERE dt = '${bizdate}'
      AND pqtid IS NOT NULL
      AND pqtid <> ''
),
daily_converted_mids AS (
    SELECT DISTINCT c.mid
    FROM daily_ad_click_events c
    JOIN daily_conversion_pqtids o
      ON c.pqtid = o.pqtid
    WHERE c.pqtid IS NOT NULL
      AND c.pqtid <> ''
),
band_user_metrics AS (
    SELECT
        u.band_order,
        COUNT(*) AS visit_uv,
        SUM(COALESCE(s.share_cnt, 0)) AS band_share_cnt,
        SUM(CASE WHEN e.mid IS NULL THEN 1 ELSE 0 END) AS no_ad_exposure_uv,
        SUM(CASE WHEN c.mid IS NOT NULL THEN 1 ELSE 0 END) AS ad_click_uv,
        SUM(CASE WHEN x.mid IS NOT NULL THEN 1 ELSE 0 END) AS converted_uv
    FROM scored_dau u
    LEFT JOIN daily_share_by_mid s
      ON u.mid = s.mid
    LEFT JOIN daily_ad_exposure_mids e
      ON u.mid = e.mid
    LEFT JOIN daily_ad_click_mids c
      ON u.mid = c.mid
    LEFT JOIN daily_converted_mids x
      ON u.mid = x.mid
    GROUP BY u.band_order
),
daily_clicks AS (
    SELECT
        shareid,
        machinecode AS return_mid
    FROM loghubods.user_share_log
    WHERE dt = '${bizdate}'
      AND topic = 'click'
      AND shareid IS NOT NULL
      AND shareid <> ''
      AND machinecode IS NOT NULL
      AND machinecode <> ''
),
daily_source_shares AS (
    SELECT DISTINCT
        s.shareid,
        u.band_order
    FROM loghubods.user_share_log s
    JOIN scored_dau u
      ON s.machinecode = u.mid
    WHERE s.dt = '${bizdate}'
      AND s.topic = 'share'
      AND s.shareid IS NOT NULL
      AND s.shareid <> ''
),
band_share_return AS (
    SELECT
        s.band_order,
        COUNT(DISTINCT c.return_mid) AS band_same_day_share_return_uv
    FROM daily_source_shares s
    JOIN daily_clicks c
      ON s.shareid = c.shareid
    GROUP BY s.band_order
),
daily_totals AS (
    SELECT
        (SELECT COUNT(*) FROM dau) AS total_dau,
        (SELECT COALESCE(SUM(share_cnt), 0) FROM daily_share_by_mid) AS total_share_cnt,
        (SELECT COUNT(DISTINCT return_mid) FROM daily_clicks) AS total_return_uv,
        (
            SELECT COUNT(*)
            FROM dau d
            LEFT JOIN daily_ad_exposure_mids e
              ON d.mid = e.mid
            WHERE e.mid IS NULL
        ) AS total_no_ad_exposure_uv,
        (SELECT COUNT(*) FROM daily_ad_click_mids) AS total_ad_click_uv,
        (SELECT COUNT(*) FROM daily_converted_mids) AS total_converted_uv,
        (
            SELECT COUNT(DISTINCT c.return_mid)
            FROM daily_source_shares s
            JOIN daily_clicks c
              ON s.shareid = c.shareid
        ) AS total_same_day_share_return_uv
)
SELECT
    '${bizdate}' AS dt,
    b.quantile_band AS `安全分分位点`,
    CASE
        WHEN b.band_order = 1 THEN CONCAT('S > ', CAST(t.p80_score AS STRING))
        WHEN b.band_order = 2 THEN CONCAT(
            CAST(t.p60_score AS STRING), ' < S <= ', CAST(t.p80_score AS STRING)
        )
        WHEN b.band_order = 3 THEN CONCAT(
            CAST(t.p40_score AS STRING), ' < S <= ', CAST(t.p60_score AS STRING)
        )
        WHEN b.band_order = 4 THEN CONCAT(
            CAST(t.p20_score AS STRING), ' < S <= ', CAST(t.p40_score AS STRING)
        )
        ELSE CONCAT('S <= ', CAST(t.p20_score AS STRING))
    END AS score,
    t.p20_score AS p20_score,
    t.p40_score AS p40_score,
    t.p60_score AS p60_score,
    t.p80_score AS p80_score,
    d.total_dau AS `DAU（当天日活总数）`,
    COALESCE(m.visit_uv, 0) AS `访问UV（每个分位点对应人数）`,
    COALESCE(m.visit_uv, 0) * 1.0 / NULLIF(d.total_dau, 0) AS `访问UV占比`,
    d.total_share_cnt AS `当日总分享次数`,
    COALESCE(m.band_share_cnt, 0) AS `分享次数（每个分位点对应次数）`,
    d.total_return_uv AS `当日总回流`,
    COALESCE(m.band_share_cnt, 0) * 1.0 / NULLIF(d.total_share_cnt, 0) AS `分享次数占比`,
    d.total_same_day_share_return_uv AS `当日分享当日回流总人数`,
    COALESCE(r.band_same_day_share_return_uv, 0) AS `当日分享当日回流（各分位点人数）`,
    COALESCE(r.band_same_day_share_return_uv, 0) * 1.0
      / NULLIF(d.total_same_day_share_return_uv, 0) AS `当日分享当日回流人数占比`,
    d.total_no_ad_exposure_uv AS `总广告无曝光人数`,
    COALESCE(m.no_ad_exposure_uv, 0) AS `无广告曝光人数`,
    COALESCE(m.no_ad_exposure_uv, 0) * 1.0
      / NULLIF(d.total_no_ad_exposure_uv, 0) AS `无广告曝光人数占比`,
    d.total_ad_click_uv AS `总点击人数`,
    COALESCE(m.ad_click_uv, 0) AS `点击人数`,
    COALESCE(m.ad_click_uv, 0) * 1.0
      / NULLIF(d.total_ad_click_uv, 0) AS `点击人数占比`,
    d.total_converted_uv AS `总已转化人数`,
    COALESCE(m.converted_uv, 0) AS `已转化人数`,
    COALESCE(m.converted_uv, 0) * 1.0
      / NULLIF(d.total_converted_uv, 0) AS `已转化人数占比`
FROM score_bands b
CROSS JOIN thresholds t
CROSS JOIN daily_totals d
LEFT JOIN band_user_metrics m
  ON b.band_order = m.band_order
LEFT JOIN band_share_return r
  ON b.band_order = r.band_order
ORDER BY b.band_order;