--********************************************************************--
-- 回流统计维度
-- is_return_n_noself, return_n_uv_noself 计算return
-- 
-- 类目维度json名称简写
-- na:name, sp:share_pv(分享pv求和), rp:return_n_pv_noself(回流pv求和)
-- ru:return_n_uv_noself(回流uv求和), mu:max_return_n_uv_noself(一次曝光最大回流uv)
--
-- mid维度json名称简写
-- s_pv:share_pv(分享pv求和), s_cnt:share_cnt(分享次数求和), 
-- r_pv:return_pv(回流pv求和), r_uv:return_uv(回流uv求和)
-- m_s_cnt:max_share_cnt(一次曝光最大分享次数), m_r_uv:max_return_uv(一次曝光最大回流uv)
-- m_s_s:max_share_seq(最大分享序列), m_r_s:max_return_seq(最大回流序列)
-- l_s_s:last_share_seq(最近分享序列), l_r_s:last_return_seq(最近回流序列)
-- c1_s:cate1_seq(merge_first_level_cate序列-回流率), c2_s:cate2_seq(merge_second_level_cate序列-回流率)
-- l1_s:label1_seq(festive_label1序列-回流率), l2_s:label2_seq(festive_label2序列-回流率)
--********************************************************************--
CREATE TABLE IF NOT EXISTS loghubods.alg_recsys_feature_user_share_return_stat
(
    mid      STRING COMMENT 'mid'
    ,feature STRING COMMENT '统计值'
)
PARTITIONED BY 
(
    dt       STRING COMMENT '日期:20240105'
    ,hh      STRING COMMENT '小时:04'
)
STORED AS ALIORC
TBLPROPERTIES ('comment' = '统计用户最近30d的分享回流数据')
LIFECYCLE 60
;

INSERT OVERWRITE TABLE loghubods.alg_recsys_feature_user_share_return_stat PARTITION (dt = '${dt}',hh = '${hh}')
WITH t_video_info AS 
(
    SELECT  vid
            ,TRIM(channel) AS channel
            ,TRIM(cate1) AS cate1
            ,TRIM(cate2) AS cate2
            ,TRIM(label1) AS label1
            ,TRIM(label2) AS label2
    FROM    (
                SELECT  CAST(vid AS BIGINT) vid
                        ,GET_JSON_OBJECT(feature,"$.channel") AS channel
                        ,GET_JSON_OBJECT(feature,"$.merge_first_level_cate") AS cate1
                        ,GET_JSON_OBJECT(feature,"$.merge_second_level_cate") AS cate2
                        ,GET_JSON_OBJECT(feature,"$.festive_label1") AS label1
                        ,GET_JSON_OBJECT(feature,"$.festive_label2") AS label2
                        ,ROW_NUMBER() OVER (PARTITION BY vid ) AS rn
                FROM    loghubods.alg_vid_feature_basic_info
                WHERE   CONCAT(dt,hh) BETWEEN TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - 2 * 3600),'YYYYMMDDHH') AND "${dt}${hh}"
            ) 
    WHERE   rn = 1
)
,t_base AS 
(
    SELECT  mid
            ,tt1.vid
            ,is_share
            ,share_cnt
            ,is_return_noself
            ,return_1_uv_noself
            ,is_return_n_noself
            ,return_n_uv_noself
            ,ts
            ,channel
            ,cate1
            ,cate2
            ,label1
            ,label2
    FROM    (
                SELECT  (CASE    WHEN uid IS NOT NULL
                                    AND LENGTH(uid) > 0
                                    AND uid != 'null' THEN uid
                                ELSE mid
                        END) AS mid
                        ,CAST(vid AS BIGINT) vid
                        ,CAST(is_share AS BIGINT) AS is_share
                        ,CAST(share_cnt AS BIGINT) AS share_cnt
                        ,CAST(is_return_noself AS BIGINT) AS is_return_noself
                        ,CAST(return_1_uv_noself AS BIGINT) AS return_1_uv_noself
                        ,CAST(is_return_n_noself AS BIGINT) AS is_return_n_noself
                        ,CAST(return_n_uv_noself AS BIGINT) AS return_n_uv_noself
                        ,CAST(ts AS BIGINT) AS ts
                FROM    loghubods.dwd_recsys_alg_exposure_base_20250108
                WHERE   CONCAT(dt,hh) BETWEEN TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - ${hours_early} * 3600),'YYYYMMDDHH') AND TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - 1 * 3600),'YYYYMMDDHH')
                AND     mid IS NOT NULL
                AND     LENGTH(mid) > 1
                AND     vid IS NOT NULL
                AND     LENGTH(vid) > 1
                AND     is_share = '1'
            ) tt1
    LEFT JOIN t_video_info tt2
    ON      tt1.vid = tt2.vid
)
,t_total AS 
(
    SELECT  mid
            ,SUM(is_share) AS share_pv
            ,SUM(share_cnt) AS share_cnt
            ,SUM(is_return_n_noself) AS return_n_pv_noself
            ,SUM(return_n_uv_noself) AS return_n_uv_noself
            ,MAX(share_cnt) AS max_share_cnt
            ,MAX(return_n_uv_noself) AS max_return_uv
    FROM    t_base
    GROUP BY mid
)
,t_max_share AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(JSON_OBJECT("id",vid,"cnt",share_cnt,"ts",ts)) AS record
                FROM    (
                            SELECT  mid
                                    ,vid
                                    ,share_cnt
                                    ,ts
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY share_cnt DESC,ts DESC ) AS rank
                            FROM    t_base
                            WHERE   share_cnt > 0
                        ) 
                WHERE   rank <= ${max_n}
            ) 
    GROUP BY mid
)
,t_max_return AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(JSON_OBJECT("id",vid,"uv",return_n_uv_noself,"ts",ts)) AS record
                FROM    (
                            SELECT  mid
                                    ,vid
                                    ,return_n_uv_noself
                                    ,ts
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY return_n_uv_noself DESC,ts DESC ) AS rank
                            FROM    t_base
                            WHERE   is_return_n_noself > 0
                            AND     return_n_uv_noself > 0
                        ) 
                WHERE   rank <= ${max_n}
            ) 
    GROUP BY mid
)
,t_last_share AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(JSON_OBJECT("id",vid,"cnt",share_cnt,"ts",ts)) AS record
                FROM    (
                            SELECT  mid
                                    ,vid
                                    ,share_cnt
                                    ,ts
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY ts DESC ) AS rank
                            FROM    t_base
                            WHERE   share_cnt > 0
                        ) 
                WHERE   rank <= ${last_n}
            ) 
    GROUP BY mid
)
,t_last_return AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(JSON_OBJECT("id",vid,"uv",return_n_uv_noself,"ts",ts)) AS record
                FROM    (
                            SELECT  mid
                                    ,vid
                                    ,return_n_uv_noself
                                    ,ts
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY ts DESC ) AS rank
                            FROM    t_base
                            WHERE   is_return_n_noself > 0
                            AND     return_n_uv_noself > 0
                        ) 
                WHERE   rank <= ${last_n}
            ) 
    GROUP BY mid
)
,t_last_1_return AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(JSON_OBJECT("id",vid,"uv",return_1_uv_noself,"ts",ts)) AS record
                FROM    (
                            SELECT  mid
                                    ,vid
                                    ,return_1_uv_noself
                                    ,ts
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY ts DESC ) AS rank
                            FROM    t_base
                            WHERE   is_return_noself > 0
                            AND     return_1_uv_noself > 0
                        ) 
                WHERE   rank <= ${last_n}
            ) 
    GROUP BY mid
)
,t_cate1 AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(
                                    JSON_OBJECT("na",cate1,"sp",share_pv,"rp",return_n_pv_noself,"ru",return_n_uv_noself,"mu",max_return_uv)
                        ) AS record
                FROM    (
                            SELECT  mid
                                    ,cate1
                                    ,share_pv
                                    ,(CASE   WHEN return_n_pv_noself > 0 THEN return_n_pv_noself
                                            ELSE NULL
                                    END) AS return_n_pv_noself
                                    ,(CASE   WHEN return_n_uv_noself > 0 THEN return_n_uv_noself
                                            ELSE NULL
                                    END) AS return_n_uv_noself
                                    ,(CASE   WHEN max_return_uv > 0 THEN max_return_uv
                                            ELSE NULL
                                    END) AS max_return_uv
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY (1.0 * return_n_uv_noself / (share_pv + ${smooth_plus})) DESC ) AS rank
                            FROM    (
                                        SELECT  mid
                                                ,cate1
                                                ,SUM(is_share) AS share_pv
                                                ,SUM(is_return_n_noself) AS return_n_pv_noself
                                                ,SUM(return_n_uv_noself) AS return_n_uv_noself
                                                ,MAX(return_n_uv_noself) AS max_return_uv
                                        FROM    t_base
                                        WHERE   cate1 IS NOT NULL
                                        AND     cate1 != 'unknown'
                                        AND     cate1 != ''
                                        GROUP BY mid
                                                 ,cate1
                                    ) 
                        ) 
                WHERE   rank <= ${cate_n}
            ) 
    GROUP BY mid
)
,t_cate2 AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(
                                    JSON_OBJECT("na",cate2,"sp",share_pv,"rp",return_n_pv_noself,"ru",return_n_uv_noself,"mu",max_return_uv)
                        ) AS record
                FROM    (
                            SELECT  mid
                                    ,cate2
                                    ,share_pv
                                    ,(CASE   WHEN return_n_pv_noself > 0 THEN return_n_pv_noself
                                            ELSE NULL
                                    END) AS return_n_pv_noself
                                    ,(CASE   WHEN return_n_uv_noself > 0 THEN return_n_uv_noself
                                            ELSE NULL
                                    END) AS return_n_uv_noself
                                    ,(CASE   WHEN max_return_uv > 0 THEN max_return_uv
                                            ELSE NULL
                                    END) AS max_return_uv
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY (1.0 * return_n_uv_noself / (share_pv + ${smooth_plus})) DESC ) AS rank
                            FROM    (
                                        SELECT  mid
                                                ,cate2
                                                ,SUM(is_share) AS share_pv
                                                ,SUM(is_return_n_noself) AS return_n_pv_noself
                                                ,SUM(return_n_uv_noself) AS return_n_uv_noself
                                                ,MAX(return_n_uv_noself) AS max_return_uv
                                        FROM    t_base
                                        WHERE   cate2 IS NOT NULL
                                        AND     cate2 != 'unknown'
                                        AND     cate2 != ''
                                        GROUP BY mid
                                                 ,cate2
                                    ) 
                        ) 
                WHERE   rank <= ${cate_n}
            ) 
    GROUP BY mid
)
,t_label1 AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(
                                    JSON_OBJECT("na",label1,"sp",share_pv,"rp",return_n_pv_noself,"ru",return_n_uv_noself,"mu",max_return_uv)
                        ) AS record
                FROM    (
                            SELECT  mid
                                    ,label1
                                    ,share_pv
                                    ,(CASE   WHEN return_n_pv_noself > 0 THEN return_n_pv_noself
                                            ELSE NULL
                                    END) AS return_n_pv_noself
                                    ,(CASE   WHEN return_n_uv_noself > 0 THEN return_n_uv_noself
                                            ELSE NULL
                                    END) AS return_n_uv_noself
                                    ,(CASE   WHEN max_return_uv > 0 THEN max_return_uv
                                            ELSE NULL
                                    END) AS max_return_uv
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY (1.0 * return_n_uv_noself / (share_pv + ${smooth_plus})) DESC ) AS rank
                            FROM    (
                                        SELECT  mid
                                                ,label1
                                                ,SUM(is_share) AS share_pv
                                                ,SUM(is_return_n_noself) AS return_n_pv_noself
                                                ,SUM(return_n_uv_noself) AS return_n_uv_noself
                                                ,MAX(return_n_uv_noself) AS max_return_uv
                                        FROM    t_base
                                        WHERE   label1 IS NOT NULL
                                        AND     label1 != 'unknown'
                                        AND     label1 != ''
                                        GROUP BY mid
                                                 ,label1
                                    ) 
                        ) 
                WHERE   rank <= ${cate_n}
            ) 
    GROUP BY mid
)
,t_label2 AS 
(
    SELECT  mid
            ,CONCAT("[",ARRAY_JOIN(COLLECT_LIST(record),","),"]") AS seq
    FROM    (
                SELECT  mid
                        ,JSON_FORMAT(
                                    JSON_OBJECT("na",label2,"sp",share_pv,"rp",return_n_pv_noself,"ru",return_n_uv_noself,"mu",max_return_uv)
                        ) AS record
                FROM    (
                            SELECT  mid
                                    ,label2
                                    ,share_pv
                                    ,(CASE   WHEN return_n_pv_noself > 0 THEN return_n_pv_noself
                                            ELSE NULL
                                    END) AS return_n_pv_noself
                                    ,(CASE   WHEN return_n_uv_noself > 0 THEN return_n_uv_noself
                                            ELSE NULL
                                    END) AS return_n_uv_noself
                                    ,(CASE   WHEN max_return_uv > 0 THEN max_return_uv
                                            ELSE NULL
                                    END) AS max_return_uv
                                    ,ROW_NUMBER() OVER (PARTITION BY mid ORDER BY (1.0 * return_n_uv_noself / (share_pv + ${smooth_plus})) DESC ) AS rank
                            FROM    (
                                        SELECT  mid
                                                ,label2
                                                ,SUM(is_share) AS share_pv
                                                ,SUM(is_return_n_noself) AS return_n_pv_noself
                                                ,SUM(return_n_uv_noself) AS return_n_uv_noself
                                                ,MAX(return_n_uv_noself) AS max_return_uv
                                        FROM    t_base
                                        WHERE   label2 IS NOT NULL
                                        AND     label2 != 'unknown'
                                        AND     label2 != ''
                                        GROUP BY mid
                                                 ,label2
                                    ) 
                        ) 
                WHERE   rank <= ${cate_n}
            ) 
    GROUP BY mid
)
,t_result AS 
(
    SELECT  mid
            ,share_pv
            ,share_cnt
            ,(CASE   WHEN return_pv > 0 THEN return_pv
                    ELSE NULL
            END) AS return_pv
            ,(CASE   WHEN return_uv > 0 THEN return_uv
                    ELSE NULL
            END) AS return_uv
            ,(CASE   WHEN max_share_cnt > 0 THEN max_share_cnt
                    ELSE NULL
            END) AS max_share_cnt
            ,(CASE   WHEN max_return_uv > 0 THEN max_return_uv
                    ELSE NULL
            END) AS max_return_uv
            ,max_share_seq
            ,max_return_seq
            ,last_share_seq
            ,last_return_seq
            ,cate1_seq
            ,cate2_seq
            ,label1_seq
            ,label2_seq
            ,last_1_return_seq
    FROM    (
                SELECT  t1.mid
                        ,t1.share_pv
                        ,t1.share_cnt
                        ,t1.return_n_pv_noself AS return_pv
                        ,t1.return_n_uv_noself AS return_uv
                        ,t1.max_share_cnt
                        ,t1.max_return_uv
                        ,t2.seq AS max_share_seq
                        ,t3.seq AS max_return_seq
                        ,t4.seq AS last_share_seq
                        ,t5.seq AS last_return_seq
                        ,t6.seq AS cate1_seq
                        ,t7.seq AS cate2_seq
                        ,t8.seq AS label1_seq
                        ,t9.seq AS label2_seq
                        ,t10.seq AS last_1_return_seq
                FROM    t_total t1
                LEFT JOIN t_max_share t2
                ON      t1.mid = t2.mid
                LEFT JOIN t_max_return t3
                ON      t1.mid = t3.mid
                LEFT JOIN t_last_share t4
                ON      t1.mid = t4.mid
                LEFT JOIN t_last_return t5
                ON      t1.mid = t5.mid
                LEFT JOIN t_cate1 t6
                ON      t1.mid = t6.mid
                LEFT JOIN t_cate2 t7
                ON      t1.mid = t7.mid
                LEFT JOIN t_label1 t8
                ON      t1.mid = t8.mid
                LEFT JOIN t_label2 t9
                ON      t1.mid = t9.mid
                LEFT JOIN t_last_1_return t10
                ON      t1.mid = t10.mid
            ) 
)SELECT  mid
        ,JSON_FORMAT(
                    JSON_OBJECT('s_pv',share_pv,"s_cnt",share_cnt,"r_pv",return_pv,"r_uv",return_uv,"m_s_cnt",max_share_cnt,"m_r_uv",max_return_uv,"m_s_s",max_share_seq,"m_r_s",max_return_seq,"l_s_s",last_share_seq,"l_r_s",last_return_seq,"c1_s",cate1_seq,"c2_s",cate2_seq,"l1_s",label1_seq,"l2_s",label2_seq,"l_r1_s",last_1_return_seq)
        ) AS feature
FROM    t_result
;