add rule rank h by 24h task

config.py

@@ -105,6 +105,15 @@ class BaseConfig(object):
         'rule2': {'cal_score_func': 2, 'return_count': 100},
     }
 
+    # 小时级更新过去24h数据
+    PROJECT_24H = 'loghubods'
+    TABLE_24H = 'video_data_each_hour_dataset_24h_total'
+
+    # 小时级更新过去24h数据规则参数
+    RULE_PARAMS_24H = {
+        'rule1': {'cal_score_func': 2, 'return_count': 100},
+    }
+
     # 地域分组小时级规则更新使用数据
     PROJECT_REGION = 'loghubods'
     TABLE_REGION = 'video_each_hour_update_province'
@@ -147,6 +156,14 @@ class BaseConfig(object):
     # 完整格式：com.weiqu.video.recall.hot.item.score.dup.day.pre.{rule_key}.{date}
     RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE = 'com.weiqu.video.recall.hot.item.score.dup.day.pre.'
 
+    # 小程序小时级24h数据更新结果存放 redis key前缀，完整格式：com.weiqu.video.recall.item.score.day.{rule_key}.{date}.{h}
+    RECALL_KEY_NAME_PREFIX_BY_24H = 'com.weiqu.video.recall.item.score.24h.'
+    # 小程序离线ROV模型结果与小程序小时级24h更新结果去重后 存放 redis key前缀，
+    # 完整格式：com.weiqu.video.recall.hot.item.score.dup.24h.{rule_key}.{date}.{h}
+    RECALL_KEY_NAME_PREFIX_DUP_24H = 'com.weiqu.video.recall.hot.item.score.dup.24h.'
+    # 小时级视频状态不符合推荐要求的列表 redis key，完整格式：com.weiqu.video.filter.h.item.24h.{rule_key}
+    H_VIDEO_FILER_24H = 'com.weiqu.video.filter.h.item.24h.'
+
     # 小程序地域分组小时级更新结果存放 redis key前缀，完整格式：com.weiqu.video.recall.item.score.region.h.{region}.{rule_key}.{date}.{h}
     RECALL_KEY_NAME_PREFIX_REGION_BY_H = 'com.weiqu.video.recall.item.score.region.h.'
     # 小程序地域分组天级更新结果与小程序地域分组小时级更新结果去重后 存放 redis key前缀，

rule_rank_h_by_24h.py

@@ -0,0 +1,211 @@
+import pandas as pd
+import math
+from odps import ODPS
+from threading import Timer
+from datetime import datetime, timedelta
+from get_data import get_data_from_odps
+from db_helper import RedisHelper
+from config import set_config
+from log import Log
+
+config_, _ = set_config()
+log_ = Log()
+
+features = [
+    'videoid',
+    'preview人数',  # 过去24h预曝光人数
+    'view人数',  # 过去24h曝光人数
+    'play人数',  # 过去24h播放人数
+    'share人数',  # 过去24h分享人数
+    '回流人数',  # 过去24h分享，过去24h回流人数
+    'preview次数',  # 过去24h预曝光次数
+    'view次数',  # 过去24h曝光次数
+    'play次数',  # 过去24h播放次数
+    'share次数',  # 过去24h分享次数
+]
+
+
+def get_rov_redis_key(now_date):
+    # 获取rov模型结果存放key
+    redis_helper = RedisHelper()
+    now_dt = datetime.strftime(now_date, '%Y%m%d')
+    key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{now_dt}'
+    if not redis_helper.key_exists(key_name=key_name):
+        pre_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
+        key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
+    return key_name
+
+
+def h_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.strftime(now_date, '%Y%m%d%H')
+        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_feature_data(now_date, project, table):
+    """获取特征数据"""
+    dt = datetime.strftime(now_date, '%Y%m%d%H')
+    # dt = '20220425'
+    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)
+    return feature_df
+
+
+def cal_score1(df):
+    # score1计算公式: score = 回流人数/(view人数+10000)
+    df = df.fillna(0)
+    df['score'] = df['回流人数'] / (df['view人数'] + 1000)
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score2(df):
+    # score2计算公式: score = share次数/(view+1000)+0.01*return/(share次数+100)
+    df = df.fillna(0)
+    df['share_rate'] = df['share次数'] / (df['view人数'] + 1000)
+    df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
+    df['score'] = df['share_rate'] + 0.01 * df['back_rate']
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def video_rank_h(df, now_date, now_h, rule_key, param):
+    """
+    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
+    :param df:
+    :param now_date:
+    :param now_h:
+    :param rule_key: 天级规则数据进入条件
+    :param param: 天级规则数据进入条件参数
+    :return:
+    """
+    # 获取rov模型结果
+    redis_helper = RedisHelper()
+    key_name = get_rov_redis_key(now_date=now_date)
+    initial_data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1, with_scores=True)
+    log_.info(f'initial data count = {len(initial_data)}')
+
+    # 获取符合进入召回源条件的视频
+    return_count = param.get('return_count')
+    if return_count:
+        day_recall_df = df[df['回流人数'] > return_count]
+    else:
+        day_recall_df = df
+    day_recall_videos = day_recall_df['videoid'].to_list()
+    log_.info(f'h_by24h_recall videos count = {len(day_recall_videos)}')
+
+    # 写入对应的redis
+    now_dt = datetime.strftime(now_date, '%Y%m%d')
+    day_video_ids = []
+    day_recall_result = {}
+    for video_id in day_recall_videos:
+        score = day_recall_df[day_recall_df['videoid'] == video_id]['score']
+        day_recall_result[int(video_id)] = float(score)
+        day_video_ids.append(int(video_id))
+    day_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{rule_key}.{now_dt}.{now_h}"
+    if len(day_recall_result) > 0:
+        redis_helper.add_data_with_zset(key_name=day_recall_key_name, data=day_recall_result, expire_time=24 * 3600)
+        # 清空线上过滤应用列表
+        redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{rule_key}")
+
+    # 去重更新rov模型结果，并另存为redis中
+    initial_data_dup = {}
+    for video_id, score in initial_data:
+        if int(video_id) not in day_video_ids:
+            initial_data_dup[int(video_id)] = score
+    log_.info(f"initial data dup count = {len(initial_data_dup)}")
+
+    initial_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_24H}{rule_key}.{now_dt}.{now_h}"
+    if len(initial_data_dup) > 0:
+        redis_helper.add_data_with_zset(key_name=initial_key_name, data=initial_data_dup, expire_time=24 * 3600)
+
+
+def rank_by_h(now_date, now_h, rule_params, project, table):
+    # 获取特征数据
+    feature_df = get_feature_data(now_date=now_date, project=project, table=table)
+    # rank
+    for key, value in rule_params.items():
+        log_.info(f"rule = {key}, param = {value}")
+        # 计算score
+        cal_score_func = value.get('cal_score_func', 1)
+        if cal_score_func == 2:
+            score_df = cal_score2(df=feature_df)
+        else:
+            score_df = cal_score1(df=feature_df)
+        video_rank_h(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value)
+        # to-csv
+        score_filename = f"score_by24h_{key}_{datetime.strftime(now_date, '%Y%m%d%H')}.csv"
+        score_df.to_csv(f'./data/{score_filename}')
+        # to-logs
+        log_.info({"date": datetime.strftime(now_date, '%Y%m%d%H'),
+                   "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_BY_24H,
+                   "rule_key": key,
+                   "score_df": score_df[['videoid', 'score']]})
+
+
+def h_rank_bottom(now_date, now_h, rule_key):
+    """未按时更新数据，用模型召回数据作为当前的数据"""
+    log_.info(f"rule_key = {rule_key}")
+    now_dt = datetime.strftime(now_date, '%Y%m%d')
+    redis_helper = RedisHelper()
+    key_name = get_rov_redis_key(now_date=now_date)
+    initial_data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1, with_scores=True)
+    final_data = dict()
+    for video_id, score in initial_data:
+        final_data[video_id] = score
+    key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_24H}{rule_key}.{now_dt}.{now_h}"
+    if len(final_data) > 0:
+        redis_helper.add_data_with_zset(key_name=key_name, data=final_data, expire_time=7 * 24 * 3600)
+        # 清空线上过滤应用列表
+        redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{rule_key}")
+
+
+def h_timer_check():
+    project = config_.PROJECT_24H
+    table = config_.TABLE_24H
+    rule_params = config_.RULE_PARAMS_24H
+    now_date = datetime.today()
+    log_.info(f"now_date: {datetime.strftime(now_date, '%Y%m%d%H')}")
+    now_min = datetime.now().minute
+    now_h = datetime.now().hour
+    # 查看当前天级更新的数据是否已准备好
+    h_data_count = h_data_check(project=project, table=table, now_date=now_date)
+    if h_data_count > 0:
+        log_.info(f'h_by24h_data_count = {h_data_count}')
+        # 数据准备好，进行更新
+        rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params, project=project, table=table)
+    elif now_min > 50:
+        log_.info('h_by24h_recall data is None!')
+        for key, _ in rule_params.items():
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_key=key)
+    else:
+        # 数据没准备好，1分钟后重新检查
+        Timer(60, h_timer_check).start()
+
+
+if __name__ == '__main__':
+    h_timer_check()

rule_rank_h_by_24h_task.sh

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

videos_filter.py

@@ -634,6 +634,49 @@ def filter_region_videos_by_day():
     log_.info("region_day videos filter end!")
 
 
+def filter_rov_h_24h():
+    """过滤小程序小时级更新24h数据"""
+    rule_params = config_.RULE_PARAMS_24H
+    log_.info("rov_h_by24h pool filter start ...")
+    redis_helper = RedisHelper()
+    # 获取当前日期
+    now_date = date.today().strftime('%Y%m%d')
+    # 获取当前所在小时
+    now_h = datetime.now().hour
+    log_.info(f'now_date = {now_date}, now_h = {now_h}.')
+    for key, value in rule_params.items():
+        log_.info(f"rule = {key}, param = {value}")
+        # 需过滤两个视频列表
+        key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_24H, config_.RECALL_KEY_NAME_PREFIX_DUP_24H]
+        for i, key_prefix in enumerate(key_prefix_list):
+            # 拼接key
+            key_name = f"{key_prefix}{key}.{now_date}.{now_h}"
+            log_.info(f"key_name: {key_name}")
+            # 获取视频
+            data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1)
+            if data is None:
+                log_.info("data is None")
+                log_.info("filter end!")
+                continue
+            # 过滤
+            video_ids = [int(video_id) for video_id in data]
+            filtered_result = filter_video_status(video_ids=video_ids)
+            # 求差集，获取需要过滤掉的视频，并从redis中移除
+            filter_videos = set(video_ids) - set(filtered_result)
+            log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
+                                                                                          len(filtered_result),
+                                                                                          len(filter_videos)))
+            if len(filter_videos) == 0:
+                log_.info("filter end!")
+                continue
+            redis_helper.remove_value_from_zset(key_name=key_name, value=list(filter_videos))
+            if i == 0:
+                # 将小时级的数据需要过滤的视频加入到线上过滤应用列表中
+                redis_helper.add_data_with_set(key_name=f"{config_.H_VIDEO_FILER_24H}{key}",
+                                               values=filter_videos, expire_time=2*3600)
+    log_.info("rov_h_by24h pool filter end!")
+
+
 def main():
     try:
         # ROV召回池视频过滤
@@ -669,6 +712,8 @@ def main():
         filter_region_videos()
         # 过滤地域分组天级视频
         filter_region_videos_by_day()
+        # 过滤小时级更新24h视频
+        filter_rov_h_24h()
     except Exception as e:
         log_.error(traceback.format_exc())
         send_msg_to_feishu(