add rule rank h

config.py

@@ -72,6 +72,8 @@ class BaseConfig(object):
 
     # 小程序离线ROV模型结果存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.{date}
     RECALL_KEY_NAME_PREFIX = 'com.weiqu.video.recall.hot.item.score.'
+    # 小程序小时级更新结果存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.h.{date}.{h}
+    RECALL_KEY_NAME_PREFIX_BY_H = 'com.weiqu.video.recall.hot.item.score.h.'
 
     # app应用 小程序离线ROV模型结果存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.app.{date}
     RECALL_KEY_NAME_PREFIX_APP = 'com.weiqu.video.recall.hot.item.score.app.'

rule_rank_h.py

@@ -0,0 +1,182 @@
+import datetime
+import pandas as pd
+import math
+from odps import ODPS
+from threading import Timer
+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()
+
+project = 'loghubods'
+table = 'video_each_hour_update'
+features = [
+    'videoid',
+    'lastonehour_view',  # 过去1小时曝光
+    'lastonehour_play',  # 过去1小时播放
+    'lastonehour_share',  # 过去1小时分享
+    'lastonehour_return',  # 过去1小时分享，过去1小时回流
+]
+
+
+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.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_rov_redis_key(now_date):
+    # 获取rov模型结果存放key
+    redis_helper = RedisHelper()
+    now_dt = datetime.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.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
+        key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
+    return key_name
+
+
+def get_feature_data(now_date):
+    """获取特征数据"""
+    dt = datetime.datetime.strftime(now_date, '%Y%m%d%H')
+    # dt = '2022041310'
+    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_score(df):
+    """
+    计算score
+    :param df: 特征数据
+    :return:
+    """
+    # score计算公式: sharerate*backrate*logback*ctr
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return/(lastonehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_view+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # score = sharerate * backrate * LOG(lastonehour_return+1) * K2
+
+    df = df.fillna(0)
+    df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000)
+    df['back_rate'] = df['lastonehour_return'] / (df['lastonehour_share'] + 10)
+    df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
+    df['ctr'] = df['lastonehour_play'] / (df['lastonehour_view'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+    df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def video_rank(df, now_date, now_h):
+    """
+    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
+    :param df:
+    :param now_date:
+    :param now_h:
+    :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)}')
+
+    # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
+    h_recall_df = df[(df['lastonehour_return'] >= 20) & (df['score'] >= 0.005)]
+    h_recall_videos = h_recall_df['videoid'].to_list()
+    print(len(h_recall_videos))
+
+    # 去重合并
+    final_videos = [int(item) for item in h_recall_videos]
+    temp_videos = [int(video_id) for video_id, _ in initial_data if int(video_id) not in final_videos]
+    final_videos = final_videos + temp_videos
+    print(len(final_videos))
+
+    # 重新给定score
+    final_data = {}
+    for i, video_id in enumerate(final_videos):
+        score = 100 - i * config_.ROV_SCORE_D
+        final_data[video_id] = score
+
+    # 存入对应的redis
+    final_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=24 * 3600)
+
+
+def rank_by_h(now_date, now_h):
+    # 获取特征数据
+    feature_df = get_feature_data(now_date=now_date)
+    # 计算score
+    score_df = cal_score(df=feature_df)
+    # rank
+    video_rank(df=score_df, now_date=now_date, now_h=now_h)
+    # to-csv
+    score_filename = f"score_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
+    score_df.to_csv(f'./data/{score_filename}')
+
+
+def h_rank_bottom(now_date, now_h):
+    """未按时更新数据，用rov模型结果作为当前小时的数据"""
+    # 获取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)
+    final_data = dict()
+    for video_id, score in initial_data:
+        final_data[video_id] = score
+    # 存入对应的redis
+    final_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=24 * 3600)
+
+
+def h_timer_check():
+    now_date = datetime.datetime.today()
+    log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}")
+    now_h = datetime.datetime.now().hour
+    now_min = datetime.datetime.now().minute
+    # 查看当前小时更新的数据是否已准备好
+    op_data_count = h_data_check(project=config_.APP_OP_PROJECT, table=config_.APP_OP_TABLE, now_date=now_date)
+    if op_data_count > 0:
+        # 数据准备好，进行更新
+        rank_by_h(now_date=now_date, now_h=now_h)
+    elif now_min > 50:
+        log_.info('op data is None, use bottom data!')
+        h_rank_bottom(now_date=now_date, now_h=now_h)
+    else:
+        # 数据没准备好，1分钟后重新检查
+        Timer(60, h_timer_check).start()
+
+
+if __name__ == '__main__':
+    # df1 = get_feature_data()
+    # res = cal_score(df=df1)
+    # video_rank(df=res, now_date=datetime.datetime.today())
+    # rank_by_h()
+    h_timer_check()

rule_rank_h_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.py
+elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
+    cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/rule_rank_h.py
+fi

videos_filter.py

@@ -365,6 +365,38 @@ def filter_app_pool():
     log_.info("app pool filter end!")
 
 
+def filter_rov_h():
+    """过滤小程序小时级数据"""
+    log_.info("rov_h 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}.')
+    # 拼接key
+    key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{now_date}.{now_h}"
+    # 获取视频
+    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("rov_h pool filter end!")
+        return
+    # 过滤
+    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("rov_h pool filter end!")
+        return
+    redis_helper.remove_value_from_zset(key_name=key_name, value=list(filter_videos))
+    log_.info("rov_h pool filter end!")
+
+
 def main():
     try:
         # ROV召回池视频过滤
@@ -386,6 +418,8 @@ def main():
         filter_position_videos()
         # 过滤票圈视频APP小时级数据
         filter_app_pool()
+        # 过滤小程序小时级数据
+        filter_rov_h()
     except Exception as e:
         log_.error(traceback.format_exc())
         send_msg_to_feishu(