-- Task: 20250104_returnv2_cf召回_rov_scenetype  ID: 1022960345  Type: ODPS_SQL
--@exclude_input=loghubods.user_share_log_flow
--@exclude_input=loghubods.video_action_log_flow_new
--odps sql 
--********************************************************************--
--author:赵海鹏
--create time:2025-01-04 17:37:45
-- 目的: 找到某个视频上回流的人，还通过哪些视频回流或者还看了哪些视频
-- 1. 表1：统计用户在过去N小时内，在某个视频上的曝光和回流数
-- 2. 表2：统计某个视频在过去N小时内分享的 不同用户的回流次数【尽量关联scenetype, 通过曝光日志关联】
-- 3. 表2作为主表，通过mid, Left Join 表1，
--********************************************************************--
CREATE TABLE IF NOT EXISTS loghubods.alg_recsys_recall_cf_mid_rov_scenetype
(
    sence_type STRING COMMENT '用户入口来源类型，[私聊 群聊 公众号文章等]。(沿用上游的拼写错误，正确是scene)'
    ,vid        STRING COMMENT '回流视频，作为触发源头。'
    ,vids       STRING COMMENT ''
    ,scores     STRING COMMENT ''
    ,ranks      STRING COMMENT ''
    ,cnt        STRING COMMENT ''
)
PARTITIONED BY 
(
    dt          STRING COMMENT '日期:20241216'
    ,hh         STRING COMMENT '小时:14'
)
STORED AS ALIORC
TBLPROPERTIES ('comment' = '模拟returnv2，通过回流找mid，mid找曝光。基于rovn的召回策略')
LIFECYCLE 90
;

INSERT OVERWRITE TABLE loghubods.alg_recsys_recall_cf_mid_rov_scenetype PARTITION (dt = '${dt}',hh = '${hh}')
WITH t_view AS 
(
    SELECT  apptype
            ,pagesource
            ,uid
            ,mid
            ,clienttimestamp
            ,sessionid
            ,subsessionid
            ,shareid
            ,headvideoid
            ,videoid AS vid
            ,CASE   WHEN hotsencetype IN ('1008','1007','1058','1074','1010') THEN hotsencetype
                    ELSE 'other'
            END AS hotsencetype
            ,CONCAT(year,month,day,hour) AS dthh
            ,GET_JSON_OBJECT(extparams,'$.recomTraceId') AS recomTraceId
    FROM    loghubods.video_action_log_flow_new
    WHERE   CONCAT(year,month,day,hour) >= TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - ${time_diff}),'YYYYMMDDHH')
    AND     CONCAT(year,month,day,hour) < '${dt}${hh}'
    AND     businesstype = 'videoView'
    AND     apptype NOT IN ("12")
)
,t_return AS -- 过去N小时内的回流                                                    
(
    SELECT  a.apptype
            ,a.vid
            ,a.mid
            ,a.shareid
            ,COALESCE(b.hotsencetype,"other") AS hotsencetype
    FROM    (
                SELECT  apptype
                        ,clickobjectid AS vid
                        ,machinecode AS mid
                        ,pagesource
                        ,recomtraceid
                        ,sessionid
                        ,shareid
                        ,subsessionid
                        ,CONCAT(year,month,day,hour) AS dthh
                        ,COALESCE(CAST(clienttimestamp AS BIGINT) / 1000,0) AS ts
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) >= TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - ${time_diff}),'YYYYMMDDHH')
                AND     CONCAT(year,month,day,hour) < '${dt}${hh}'
                AND     __topic__ = "click"
                AND     apptype NOT IN ("12")
                AND     pagesource REGEXP "-pages/user-videos-share$"
            ) a
    LEFT JOIN   (
                    SELECT  *
                    FROM    (
                                SELECT  apptype
                                        ,mid
                                        ,vid
                                        ,subsessionid
                                        ,pagesource
                                        ,hotsencetype
                                        ,ROW_NUMBER() OVER (PARTITION BY apptype,mid,vid,subsessionid,pagesource ORDER BY CAST(clienttimestamp AS BIGINT) DESC ) AS rn
                                FROM    t_view
                                WHERE   pagesource REGEXP "-pages/user-videos-share$"
                            ) 
                    WHERE   rn = 1
                ) b
    ON      a.apptype = b.apptype
    AND     a.mid = b.mid
    AND     a.vid = b.vid
    AND     a.subsessionid = b.subsessionid
    AND     a.pagesource = b.pagesource
)
,t_share AS -- 查询分享记录，多次分享保留最近的一次                                                                               
(
    SELECT  *
    FROM    (
                SELECT  apptype
                        ,shareobjectid AS vid
                        ,shareid
                        ,machinecode AS mid
                        ,COALESCE(CAST(clienttimestamp AS BIGINT) / 1000,0) AS share_ts
                        ,ROW_NUMBER() OVER (PARTITION BY apptype,shareobjectid,shareid,machinecode ORDER BY CAST(clienttimestamp AS BIGINT) DESC ) AS rn
                FROM    loghubods.user_share_log_flow
                WHERE   CONCAT(year,month,day,hour) >= TO_CHAR(FROM_UNIXTIME(UNIX_TIMESTAMP(TO_DATE('${dt}${hh}','YYYYMMDDHH')) - ${time_diff}),'YYYYMMDDHH')
                AND     CONCAT(year,month,day,hour) < '${dt}${hh}'
                AND     __topic__ = "share"
                AND     apptype NOT IN ("12")
                AND     shareobjectid IS NOT NULL
                AND     shareobjectid != ''
            ) 
    WHERE   rn = 1
)
,t_left AS -- 最近N小时内回流，且回流对应的分享也是最近N小时内                                                                        
(
    SELECT  tr.vid
            ,tr.mid
            ,tr.hotsencetype
            ,COUNT(1) AS cnt
    FROM    t_return tr
    LEFT JOIN t_share ts
    ON      tr.apptype = ts.apptype
    AND     tr.shareid = ts.shareid
    AND     tr.vid = ts.vid
    AND     tr.mid != ts.mid
    WHERE   ts.share_ts IS NOT NULL
    GROUP BY tr.vid
             ,tr.mid
             ,tr.hotsencetype
)
,t_right AS -- 查询最近N小时内，某个用户在某个视频上的曝光次数和回流次数                                                            
(
    SELECT  tv.mid
            ,tv.vid
            ,COALESCE(tv.view_cnt,0) AS view_cnt
            ,COALESCE(tr.return_cnt,0) AS return_cnt
    FROM    (
                SELECT  mid
                        ,vid
                        ,COUNT(1) AS view_cnt
                FROM    t_view
                GROUP BY mid
                         ,vid
            ) tv
    LEFT JOIN   (
                    SELECT  ts.mid
                            ,ts.vid
                            ,COUNT(tr.mid) AS return_cnt
                    FROM    t_share ts
                    LEFT JOIN t_return tr
                    ON      ts.shareid = tr.shareid
                    AND     ts.vid = tr.vid
                    GROUP BY ts.mid
                             ,ts.vid
                ) tr
    ON      tr.mid = tv.mid
    AND     tr.vid = tv.vid
)
,t_va_vb_return_view AS 
(
    SELECT  a.vid AS vid_left
            ,b.vid AS vid_right
            ,a.hotsencetype
            ,COUNT(DISTINCT a.mid) AS va_vb_together_cnt
    FROM    t_left a
    JOIN    (
                SELECT  DISTINCT vid
                        ,mid
                FROM    t_view
            ) b
    ON      a.mid = b.mid
    GROUP BY a.vid
             ,b.vid
             ,a.hotsencetype
)
,t_result_v1 AS 
(
    SELECT  a.vid AS vid_left
            ,b.vid AS vid_right
            ,a.hotsencetype
            ,SUM(b.view_cnt) AS view
            ,SUM(IF(b.return_cnt > 0,1,0)) AS return
            ,SUM(IF(b.return_cnt > 0,1,0)) / SUM(b.view_cnt) * LN(SUM(IF(b.return_cnt > 0,1,0)) + 1) AS score
    FROM    (
                SELECT  b.vid
                        ,b.mid
                        ,b.hotsencetype
                FROM    (
                            SELECT  vid
                            FROM    t_left
                            GROUP BY vid
                            HAVING  COUNT(DISTINCT mid) >= 5
                        ) a
                JOIN    t_left b
                ON      a.vid = b.vid
            ) a
    JOIN    t_right b
    ON      a.mid = b.mid
    AND     a.vid <> b.vid
    GROUP BY a.vid
             ,b.vid
             ,a.hotsencetype
)
,t_result_v2 AS 
(
    SELECT  a.vid_left
            ,a.vid_right
            ,a.hotsencetype
            ,a.view
            ,a.return
            ,a.score
            ,b.va_vb_together_cnt
            ,ROW_NUMBER() OVER (PARTITION BY a.vid_left ORDER BY b.va_vb_together_cnt DESC ) AS rn_old
    FROM    t_result_v1 a
    JOIN    t_va_vb_return_view b
    ON      a.vid_left = b.vid_left
    AND     a.vid_right = b.vid_right
    AND     a.hotsencetype = b.hotsencetype
)
,t_recall AS 
(
    SELECT  *
    FROM    (
                SELECT  hotsencetype
                        ,vid_left
                        ,vid_right
                        ,view
                        ,return
                        ,score
                        ,ROW_NUMBER() OVER (PARTITION BY hotsencetype,vid_left ORDER BY score DESC ) AS rn
                FROM    t_result_v2
                WHERE   score > 0
                AND     rn_old <= 100
            ) 
    WHERE   rn <= 100
    DISTRIBUTE BY hotsencetype,vid_left
    SORT BY hotsencetype
,vid_left
,score DESC
)SELECT  hotsencetype as sence_type
        ,vid_left AS vid
        ,CONCAT_WS(',',COLLECT_LIST(vid_right)) AS vids
        ,CONCAT_WS(',',COLLECT_LIST(CAST(round(score,4) AS STRING))) AS scores
        ,CONCAT_WS(',',COLLECT_LIST(CAST(rn AS STRING))) AS ranks
        ,COUNT(vid_right) AS cnt
FROM    t_recall
GROUP BY hotsencetype
         ,vid_left
ORDER BY COUNT(vid_right) DESC
;