Merge branch 'religion-abtest-update-20221216'

config.py

@@ -374,8 +374,36 @@ class BaseConfig(object):
         ],
     }
 
+    # 宗教视频更新使用数据
+    RELIGION_VIDEOS = {
+        'catholicism': {
+            'project': 'loghubods',
+            'table': 'catholicism_videolist',
+            # 视频列表更新结果存放 redis key 前缀，完整格式：'religion:catholicism:videos:item:{date}'
+            'key_name_prefix': 'religion:catholicism:videos:item:',
+        },  # 天主教
+
+        'christianity': {
+            'project': 'loghubods',
+            'table': 'christianity_videolist',
+            # 视频列表更新结果存放 redis key 前缀，完整格式：'religion:christianity:videos:item:{date}'
+            'key_name_prefix': 'religion:christianity:videos:item:',
+        },  # 基督教
+    }
+
     # 老好看宗教实验数据
-    LHK_RULE_PARAMS = {'data': 'lhk_data', 'rule': 'lhk_rule'}
+    LHK_RULE_PARAMS = {
+        'rule_params': {
+            'lhk_rule1': {'initial_param': {'data': 'data1', 'rule': 'rule4-1'},
+                          'religion_name': 'catholicism', 'rank_count': 2},
+            'lhk_rule2': {'initial_param': {'data': 'data1', 'rule': 'rule4-1'},
+                          'religion_name': 'christianity', 'rank_count': 2},
+        },
+        'params_list': [
+            {'data': 'lhk_data', 'rule': 'lhk_rule1'},  # 天主教-2插1
+            {'data': 'lhk_data', 'rule': 'lhk_rule2'},  # 基督教-2插1
+        ]
+    }
 
     # 不区分地域数据使用相对48h数据
     RULE_PARAMS_REGION_APP_TYPE_48H = {

laohaokan_recommend_update.py

@@ -1,4 +1,5 @@
 import datetime
+import traceback
 import gevent
 from db_helper import RedisHelper
 from utils import send_msg_to_feishu
@@ -7,24 +8,25 @@ from log import Log
 
 config_, env = set_config()
 log_ = Log()
-initial_param = {'data': 'data1', 'rule': 'rule4'}
-new_param = config_.LHK_RULE_PARAMS
+# initial_param = {'data': 'data1', 'rule': 'rule4'}
+# new_param = config_.LHK_RULE_PARAMS
 redis_helper = RedisHelper()
 
 
-def get_religion_videos(now_date):
+def get_religion_videos(now_date, religion_name):
     """获取宗教视频列表"""
-    religion_key_name = f"{config_.KEY_NAME_PREFIX_RELIGION_VIDEOS}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
+    key_name_prefix = config_.RELIGION_VIDEOS[religion_name].get('key_name_prefix')
+    religion_key_name = f"{key_name_prefix}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
     if not redis_helper.key_exists(religion_key_name):
         redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d')
-        religion_key_name = f"{config_.KEY_NAME_PREFIX_RELIGION_VIDEOS}{redis_dt}"
+        religion_key_name = f"{key_name_prefix}{redis_dt}"
     religion_videos = redis_helper.get_all_data_from_zset(key_name=religion_key_name, desc=True, with_scores=True)
     if religion_videos is None:
         return []
     return religion_videos
 
 
-def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_list):
+def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_list, rank_count):
     initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True)
     if initial_data is None or len(initial_data) == 0:
         return now_videos, religion_video_id_list
@@ -39,7 +41,7 @@ def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_
         for i, video_id in enumerate(initial_video_ids):
             new_video_ids.append(video_id)
             now_videos.append(video_id)
-            if i % 2 == 1 and len(religion_video_id_list) > 0:
+            if i % rank_count == 1 and len(religion_video_id_list) > 0:
                 new_video_ids.append(religion_video_id_list[0])
                 now_videos.append(religion_video_id_list[0])
                 religion_video_id_list = religion_video_id_list[1:]
@@ -55,62 +57,73 @@ def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_
     return now_videos, religion_video_id_list
 
 
-def merge_with_region(now_date, now_h, region, religion_video_id_list):
+def merge_with_region(now_date, now_h, region, religion_video_id_list,
+                      initial_param, lhk_data_key, lhk_rule_key, rank_count):
     initial_data_key = initial_param.get('data')
     initial_rule_key = initial_param.get('rule')
-    new_data_key = new_param.get('data')
-    new_rule_key = new_param.get('rule')
     now_videos = []
     # 地域小时级数据合并
     region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{initial_data_key}:{initial_rule_key}:" \
                         f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
-    new_region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{new_data_key}:{new_rule_key}:" \
+    new_region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{lhk_data_key}:{lhk_rule_key}:" \
                             f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
     now_videos, religion_video_id_list = merge_process(initial_key_name=region_h_key_name,
                                                        new_key_name=new_region_h_key_name,
                                                        now_videos=now_videos,
-                                                       religion_video_id_list=religion_video_id_list)
+                                                       religion_video_id_list=religion_video_id_list,
+                                                       rank_count=rank_count)
     # 地域24h数据合并
     region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{initial_data_key}:" \
                           f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
-    new_region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{new_data_key}:" \
-                              f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    new_region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{lhk_data_key}:" \
+                              f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
     now_videos, religion_video_id_list = merge_process(initial_key_name=region_24h_key_name,
                                                        new_key_name=new_region_24h_key_name,
                                                        now_videos=now_videos,
-                                                       religion_video_id_list=religion_video_id_list)
+                                                       religion_video_id_list=religion_video_id_list,
+                                                       rank_count=rank_count)
 
     # 24h筛选数据合并
     dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{initial_data_key}:" \
                         f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
-    new_dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{new_data_key}:" \
-                            f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    new_dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{lhk_data_key}:" \
+                            f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
     now_videos, religion_video_id_list = merge_process(initial_key_name=dup2_24h_key_name,
                                                        new_key_name=new_dup2_24h_key_name,
                                                        now_videos=now_videos,
-                                                       religion_video_id_list=religion_video_id_list)
+                                                       religion_video_id_list=religion_video_id_list,
+                                                       rank_count=rank_count)
 
     # 24h筛选后剩余数据合并
     dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{initial_data_key}:" \
                         f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
-    new_dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{new_data_key}:" \
-                            f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    new_dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{lhk_data_key}:" \
+                            f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
     now_videos, religion_video_id_list = merge_process(initial_key_name=dup3_24h_key_name,
                                                        new_key_name=new_dup3_24h_key_name,
                                                        now_videos=now_videos,
-                                                       religion_video_id_list=religion_video_id_list)
+                                                       religion_video_id_list=religion_video_id_list,
+                                                       rank_count=rank_count)
     log_.info(f"region = {region} update end!")
 
 
-def merge_videos(now_date, now_h):
+def merge_videos(now_date, now_h, param, rule_params):
     """将宗教视频插入到默认视频列表中"""
     # 获取宗教视频列表
-    religion_videos = get_religion_videos(now_date=now_date)
+    lhk_data_key = param.get('data')
+    lhk_rule_key = param.get('rule')
+    religion_name = rule_params.get(lhk_rule_key).get('religion_name')
+    initial_param = rule_params.get(lhk_rule_key).get('initial_param')
+    rank_count = rule_params.get(lhk_rule_key).get('rank_count')
+    religion_videos = get_religion_videos(now_date=now_date, religion_name=religion_name)
     religion_video_id_list = [int(video_id) for video_id, _ in religion_videos]
     # 列表合并
     region_code_list = [code for region, code in config_.REGION_CODE.items()]
     task_list = [
-        gevent.spawn(merge_with_region, now_date, now_h, region, religion_video_id_list)
+        gevent.spawn(
+            merge_with_region,
+            now_date, now_h, region, religion_video_id_list, initial_param, lhk_data_key, lhk_rule_key, rank_count
+        )
         for region in region_code_list
     ]
     gevent.joinall(task_list)
@@ -119,25 +132,46 @@ def merge_videos(now_date, now_h):
         t = [
             gevent.spawn(
                 merge_with_region,
-                now_date, now_h, city_code, religion_video_id_list
+                now_date, now_h, city_code, religion_video_id_list, initial_param, lhk_data_key, lhk_rule_key, rank_count
             )
             for city_code in city_list
         ]
         gevent.joinall(t)
 
 
+def main():
+    try:
+        log_.info(f"laohaokan recommend data update start...")
+        now_date = datetime.datetime.today()
+        now_h = datetime.datetime.now().hour
+        log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
+        lhk_rule_params = config_.LHK_RULE_PARAMS
+        rule_params = lhk_rule_params.get('rule_params', {})
+        params_list = lhk_rule_params.get('params_list', [])
+        for param in params_list:
+            log_.info(f"param = {param} update start...")
+            merge_videos(now_date, now_h, param, rule_params)
+            log_.info(f"param = {param} update end!")
+
+        log_.info(f"laohaokan recommend data update end!")
+        send_msg_to_feishu(
+            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
+            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
+            msg_text=f"rov-offline{config_.ENV_TEXT} - 老好看推荐视频数据更新完成\n"
+                     f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}\n"
+                     f"now_h: {now_h}\n"
+                     f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}"
+        )
+    except Exception as e:
+        log_.error(f"老好看推荐视频数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
+        send_msg_to_feishu(
+            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
+            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
+            msg_text=f"rov-offline{config_.ENV_TEXT} - 老好看推荐视频数据更新失败\n"
+                     f"exception: {e}\n"
+                     f"traceback: {traceback.format_exc()}"
+        )
+
+
 if __name__ == '__main__':
-    log_.info(f"laohaokan recommend data update start...")
-    now_date = datetime.datetime.today()
-    now_h = datetime.datetime.now().hour
-    log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
-    merge_videos(now_date, now_h)
-    log_.info(f"laohaokan recommend data update end!")
-    send_msg_to_feishu(
-        webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
-        key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
-        msg_text=f"rov-offline{config_.ENV_TEXT} - 老好看推荐视频数据更新完成\n"
-                 f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}\n"
-                 f"now_h: {now_h}\n"
-                 f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}"
-    )
+    main()

region_rule_rank_h_task.sh

@@ -4,10 +4,10 @@ if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
     cd /data2/rov-offline && /root/anaconda3/bin/python /data2/rov-offline/rule_rank_h_by_24h.py &&
      /root/anaconda3/bin/python /data2/rov-offline/region_rule_rank_h_by24h.py &&
       /root/anaconda3/bin/python /data2/rov-offline/region_rule_rank_h.py '24h'
-#      /root/anaconda3/bin/python /data2/rov-offline/laohaokan_recommend_update.py
+      /root/anaconda3/bin/python /data2/rov-offline/laohaokan_recommend_update.py
 elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
     cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/rule_rank_h_by_24h.py &&
      /root/anaconda3/bin/python /data/rov-offline/region_rule_rank_h_by24h.py &&
       /root/anaconda3/bin/python /data/rov-offline/region_rule_rank_h.py '24h'
-#      /root/anaconda3/bin/python /data/rov-offline/laohaokan_recommend_update.py
+      /root/anaconda3/bin/python /data/rov-offline/laohaokan_recommend_update.py
 fi

religion_class_videos_update.py

@@ -0,0 +1,105 @@
+import datetime
+import sys
+import traceback
+import pandas as pd
+from odps import ODPS
+from threading import Timer
+from get_data import get_data_from_odps
+from utils import send_msg_to_feishu
+from db_helper import RedisHelper
+from config import set_config
+from log import Log
+
+config_, env = set_config()
+log_ = Log()
+features = ['videoid', 'play_count_total', 'gmt_create']
+
+
+def data_check(project, table, now_date):
+    """检查数据是否准备好"""
+    odps = ODPS(
+        access_id=config_.ODPS_CONFIG['ACCESSID'],
+        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
+        project=project,
+        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
+        connect_timeout=3000,
+        read_timeout=500000,
+        pool_maxsize=1000,
+        pool_connections=1000
+    )
+
+    try:
+        dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+        sql = f'select * from {project}.{table} where dt = {dt}'
+        with odps.execute_sql(sql=sql).open_reader() as reader:
+            data_count = reader.count
+    except Exception as e:
+        data_count = 0
+    return data_count
+
+
+def get_religion_videos(now_date, project, table, key_name_prefix):
+    """获取宗教视频列表"""
+    # 获取videoId
+    dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+    records = get_data_from_odps(date=dt, project=project, table=table)
+    feature_data = []
+    for record in records:
+        item = {}
+        for feature_name in features:
+            item[feature_name] = record[feature_name]
+        feature_data.append(item)
+    feature_df = pd.DataFrame(feature_data)
+    # 按照发布时间和播放量进行倒序
+    feature_df = feature_df.sort_values(by=['gmt_create', 'play_count_total'], ascending=False)
+    print(feature_df)
+    video_id_list = feature_df['videoid'].to_list()
+    # 按照排序给定分数
+    final_result = {}
+    step = 100 / (len(video_id_list) * 2)
+    for i, video_id in enumerate(video_id_list):
+        score = 100 - i * step
+        final_result[int(video_id)] = score
+
+    # 写入对应的redis
+    key_name = \
+        f"{key_name_prefix}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
+    if len(final_result) > 0:
+        redis_helper = RedisHelper()
+        redis_helper.add_data_with_zset(key_name=key_name, data=final_result, expire_time=2 * 24 * 3600)
+
+
+def timer_check(religion_name):
+    project = config_.RELIGION_VIDEOS[religion_name]['project']
+    table = config_.RELIGION_VIDEOS[religion_name]['table']
+    key_name_prefix = config_.RELIGION_VIDEOS[religion_name]['key_name_prefix']
+    now_date = datetime.datetime.today()
+    log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}")
+    # 查看当天更新的数据是否已准备好
+    data_count = data_check(project=project, table=table, now_date=now_date)
+    if data_count > 0:
+        log_.info(f'religion_name = {religion_name}, religion_videos_count = {data_count}')
+        # 数据准备好，进行更新
+        get_religion_videos(now_date=now_date, project=project, table=table, key_name_prefix=key_name_prefix)
+    else:
+        # 数据没准备好，1分钟后重新检查
+        Timer(5 * 60, timer_check, args=[religion_name]).start()
+
+
+def main():
+    try:
+        religion_name = sys.argv[1]
+        timer_check(religion_name)
+    except Exception as e:
+        log_.error(f"宗教视频数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
+        send_msg_to_feishu(
+            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
+            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
+            msg_text=f"rov-offline{config_.ENV_TEXT} - 宗教视频数据更新失败\n"
+                     f"exception: {e}\n"
+                     f"traceback: {traceback.format_exc()}"
+        )
+
+
+if __name__ == '__main__':
+    main()

religion_class_videos_update_task.sh

@@ -0,0 +1,11 @@
+source /etc/profile
+echo $ROV_OFFLINE_ENV
+if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
+    cd /data2/rov-offline &&
+    nohup /root/anaconda3/bin/python /data2/rov-offline/religion_class_videos_update.py 'catholicism' &
+    nohup /root/anaconda3/bin/python /data2/rov-offline/religion_class_videos_update.py 'christianity' &
+elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
+    cd /data/rov-offline &&
+    nohup /root/anaconda3/bin/python /data2/rov-offline/religion_class_videos_update.py 'catholicism' &
+    nohup /root/anaconda3/bin/python /data2/rov-offline/religion_class_videos_update.py 'christianity' &
+fi