增加脚本输出log功能

alg_recsys_recall_1h_region.py

@@ -0,0 +1,1272 @@
+# -*- coding: utf-8 -*-
+import multiprocessing
+import sys
+import traceback
+import gevent
+import datetime
+import pandas as pd
+import math
+from functools import reduce
+from odps import ODPS
+from threading import Timer, Thread
+from utils import MysqlHelper, RedisHelper, get_data_from_odps, filter_video_status, filter_shield_video, \
+    check_table_partition_exits, filter_video_status_app, send_msg_to_feishu, filter_political_videos
+from config import set_config
+from log import Log
+from check_video_limit_distribute import update_limit_video_score
+
+# os.environ['NUMEXPR_MAX_THREADS'] = '16'
+
+config_, _ = set_config()
+log_ = Log()
+
+region_code = config_.REGION_CODE
+
+
+RULE_PARAMS = {
+    'rule_params': {
+        'rule66': {
+            'view_type': 'video-show-region', 'platform_return_rate': 0.001,
+            'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66'
+        },
+    },
+    'data_params': config_.DATA_PARAMS,
+    'params_list': [
+        # 532
+        {'data': 'data66', 'rule': 'rule66'},
+    ],
+}
+
+features = [
+    'apptype',
+    'code',
+    'videoid',
+    'lastonehour_preview',  # 过去1小时预曝光人数 - 区分地域
+    'lastonehour_view',  # 过去1小时曝光人数 - 区分地域
+    'lastonehour_play',  # 过去1小时播放人数 - 区分地域
+    'lastonehour_share',  # 过去1小时分享人数 - 区分地域
+    'lastonehour_return',  # 过去1小时分享，过去1小时回流人数 - 区分地域
+    'lastonehour_preview_total',  # 过去1小时预曝光次数 - 区分地域
+    'lastonehour_view_total',  # 过去1小时曝光次数 - 区分地域
+    'lastonehour_play_total',  # 过去1小时播放次数 - 区分地域
+    'lastonehour_share_total',  # 过去1小时分享次数 - 区分地域
+    'platform_return',
+    'lastonehour_show',  # 不区分地域
+    'lastonehour_show_region',  # 地域分组
+    'lasttwohour_share',  # h-2小时分享人数
+    'lasttwohour_return_now',  # h-2分享，过去1小时回流人数
+    'lasttwohour_return',  # h-2分享，h-2回流人数
+    'lastthreehour_share',  # h-3小时分享人数
+    'lastthreehour_return_now',  # h-3分享，过去1小时回流人数
+    'lastthreehour_return',  # h-3分享，h-3回流人数
+
+    'lastonehour_return_new',  # 过去1小时分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+    'lasttwohour_return_now_new',  # h-2分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+    'lasttwohour_return_new',  # h-2分享，h-2回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+    'lastthreehour_return_now_new',  # h-3分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+    'lastthreehour_return_new',  # h-3分享，h-3回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+    'platform_return_new',  # 平台分发回流（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域）
+]
+
+
+def get_region_code(region):
+    """获取省份对应的code"""
+    mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO)
+    sql = f"SELECT ad_code FROM region_adcode WHERE parent_id = 0 AND region LIKE '{region}%';"
+    ad_code = mysql_helper.get_data(sql=sql)
+    return ad_code[0][0]
+
+
+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')
+        check_res = check_table_partition_exits(date=dt, project=project, table=table)
+        if check_res:
+            sql = f'select * from {project}.{table} where dt = {dt}'
+            with odps.execute_sql(sql=sql).open_reader() as reader:
+                data_count = reader.count
+        else:
+            data_count = 0
+    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_day_30day_videos(now_date, data_key, rule_key):
+    """获取天级更新相对30天的视频id"""
+    redis_helper = RedisHelper()
+    day_30day_recall_key_prefix = config_.RECALL_KEY_NAME_PREFIX_30DAY
+    now_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+    day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{now_dt}"
+    if not redis_helper.key_exists(key_name=day_30day_recall_key_name):
+        redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d')
+        day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{redis_dt}"
+    data = redis_helper.get_all_data_from_zset(key_name=day_30day_recall_key_name, with_scores=True)
+    if data is None:
+        return None
+    video_ids = [int(video_id) for video_id, _ in data]
+    return video_ids
+
+
+def get_feature_data(project, table, 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_initial(df, param):
+    """
+    计算score
+    :param df: 特征数据
+    :param param: 规则参数
+    :return:
+    """
+    # score计算公式: sharerate*backrate*logback*ctr
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return/(lastonehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+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)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score1'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
+
+    click_score_rate = param.get('click_score_rate', None)
+    back_score_rate = param.get('click_score_rate', None)
+    if click_score_rate is not None:
+        df['score'] = (1 - click_score_rate) * df['score1'] + click_score_rate * df['K2']
+    elif back_score_rate is not None:
+        df['score'] = (1 - back_score_rate) * df['score1'] + back_score_rate * df['back_rate']
+    else:
+        df['score'] = df['score1']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_add_return(df, param):
+    # score计算公式: sharerate*(backrate*logback + backrate2*logback_now2 + backrate3*logback_now3)*ctr
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return/(lastonehour_share+10)
+    # backrate2 = lasttwohour_return_now/(lasttwohour_share+10)
+    # backrate3 = lastthreehour_return_now/(lastthreehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # score = k2 * sharerate * (backrate * LOG(lastonehour_return+1) + backrate_2 * LOG(lasttwohour_return_now+1) + backrate_3 * LOG(lastthreehour_return_now+1))
+
+    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['back_rate2'] = df['lasttwohour_return_now'] / (df['lasttwohour_share'] + 10)
+    df['log_back2'] = (df['lasttwohour_return_now'] + 1).apply(math.log)
+    df['back_rate3'] = df['lastthreehour_return_now'] / (df['lastthreehour_share'] + 10)
+    df['log_back3'] = (df['lastthreehour_return_now'] + 1).apply(math.log)
+
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score'] = df['K2'] * df['share_rate'] * (
+            df['back_rate'] * df['log_back'] +
+            df['back_rate2'] * df['log_back2'] +
+            df['back_rate3'] * df['log_back3']
+    )
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_multiply_return_retention(df, param):
+    # score计算公式: k2 * sharerate * backrate * LOG(lastonehour_return+1) * 前两小时回流留存
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return/(lastonehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # 前两小时回流留存 return_retention_initial = (lasttwohour_return_now + lastthreehour_return_now)/(lasttwohour_return + lastthreehour_return + 1)
+    # return_retention = 0.5 if return_retention_initial == 0 else return_retention_initial
+    # score = k2 * sharerate * backrate * LOG(lastonehour_return+1) * return_retention
+
+    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)
+
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['return_retention_initial'] = (df['lasttwohour_return_now'] + df['lastthreehour_return_now']) / \
+                                     (df['lasttwohour_return'] + df['lastthreehour_return'] + 1)
+    df['return_retention'] = df['return_retention_initial'].apply(lambda x: 0.5 if x == 0 else x)
+
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score'] = df['K2'] * df['share_rate'] * df['back_rate'] * df['log_back'] * df['return_retention']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_update_backrate(df, param):
+    # score计算公式: k2 * sharerate * (backrate + backrate * backrate_2 * backrate_3) * LOG(lastonehour_return+1)
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return/(lastonehour_share+10)
+    # backrate2 = lasttwohour_return_now/(lasttwohour_share+10)
+    # backrate3 = lastthreehour_return_now/(lastthreehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # backrate1_3_initial = backrate * backrate_2 * backrate_3
+    # backrate1_3 = 0.02 if backrate1_3_initial == 0 else backrate1_3_initial
+    # score = k2 * sharerate * (backrate + backrate1_3) * LOG(lastonehour_return+1)
+
+    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['back_rate2'] = df['lasttwohour_return_now'] / (df['lasttwohour_share'] + 10)
+    df['back_rate3'] = df['lastthreehour_return_now'] / (df['lastthreehour_share'] + 10)
+    df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
+
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['backrate1_3_initial'] = df['back_rate'] * df['back_rate2'] * df['back_rate3']
+    df['backrate1_3'] = df['backrate1_3_initial'].apply(lambda x: 0.02 if x == 0 else x)
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score'] = df['K2'] * df['share_rate'] * (df['back_rate'] + df['backrate1_3']) * df['log_back']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_with_new_return(df, param):
+    # 回流数据使用 分享限制地域，回流不限制地域 统计数据
+    # score计算公式: sharerate*backrate*logback*ctr
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return_new/(lastonehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # score = sharerate * backrate * LOG(lastonehour_return_new+1) * K2
+
+    df = df.fillna(0)
+    df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000)
+    df['back_rate'] = df['lastonehour_return_new'] / (df['lastonehour_share'] + 10)
+    df['log_back'] = (df['lastonehour_return_new'] + 1).apply(math.log)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+    df['platform_return_rate'] = df['platform_return_new'] / df['lastonehour_return_new']
+
+    df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_multiply_return_retention_with_new_return(df, param):
+    # 回流数据使用 分享限制地域，回流不限制地域 统计数据
+    # score计算公式: k2 * sharerate * backrate * LOG(lastonehour_return_new+1) * 前两小时回流留存
+    # sharerate = lastonehour_share/(lastonehour_play+1000)
+    # backrate = lastonehour_return_new/(lastonehour_share+10)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # 前两小时回流留存 return_retention_initial = (lasttwohour_return_now_new + lastthreehour_return_now_new)/(lasttwohour_return_new + lastthreehour_return_new + 1)
+    # return_retention = 0.5 if return_retention_initial == 0 else return_retention_initial
+    # score = k2 * sharerate * backrate * LOG(lastonehour_return_new+1) * return_retention
+
+    df = df.fillna(0)
+    df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000)
+    df['back_rate'] = df['lastonehour_return_new'] / (df['lastonehour_share'] + 10)
+    df['log_back'] = (df['lastonehour_return_new'] + 1).apply(math.log)
+
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['return_retention_initial'] = (df['lasttwohour_return_now_new'] + df['lastthreehour_return_now_new']) / \
+                                     (df['lasttwohour_return_new'] + df['lastthreehour_return_new'] + 1)
+    df['return_retention'] = df['return_retention_initial'].apply(lambda x: 0.5 if x == 0 else x)
+
+    df['platform_return_rate'] = df['platform_return_new'] / df['lastonehour_return_new']
+
+    df['score'] = df['K2'] * df['share_rate'] * df['back_rate'] * df['log_back'] * df['return_retention']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_with_back_view0(df, param):
+    # score = sharerate*backrate*log(return+1)*CTR,
+    # sharerate=(lastonehour_share+1)/(lastonehour_play+1000)
+    # backrate=(lastonehour_return+1)/(lastonehour_share+10)
+    # CTR=(lastonehour_play+1)/(lastonehour_view+100), ctr不进行校正
+    df = df.fillna(0)
+    df['share_rate'] = (df['lastonehour_share'] + 1) / (df['lastonehour_play'] + 1000)
+    df['back_rate'] = (df['lastonehour_return'] + 1) / (df['lastonehour_share'] + 10)
+    df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
+    df['ctr'] = (df['lastonehour_play'] + 1) / (df['lastonehour_view'] + 100)
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+    df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['ctr']
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_with_back_view1(df, param):
+    # score = back_play_rate*log(return+1)*CTR,
+    # back_play_rate=(lastonehour_return+1)/(lastonehour_play+1000)
+    # CTR=(lastonehour_play+1)/(lastonehour_view+100), ctr不进行校正
+    df = df.fillna(0)
+    df['back_play_rate'] = (df['lastonehour_return'] + 1) / (df['lastonehour_play'] + 1000)
+    df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
+    df['ctr'] = (df['lastonehour_play'] + 1) / (df['lastonehour_view'] + 100)
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+    df['score'] = df['back_play_rate'] * df['log_back'] * df['ctr']
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_with_back_rate_exponential_weighting1(df, param):
+    """
+    计算score
+    :param df: 特征数据
+    :param param: 规则参数
+    :return:
+    """
+    # score计算公式: score = sharerate * backrate ^ 2 * LOG(lastonehour_return + 1) * K2
+    # sharerate = lastonehour_share / (lastonehour_play + 1000)
+    # backrate = lastonehour_return / (lastonehour_share + 10)
+    # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+
+    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)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score'] = df['share_rate'] * df['back_rate'] ** 2 * df['log_back'] * df['K2']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score_with_back_rate_exponential_weighting2(df, param):
+    """
+    计算score
+    :param df: 特征数据
+    :param param: 规则参数
+    :return:
+    """
+    # score计算公式: score = sharerate ^ 0.5 * backrate ^ 2 * LOG(lastonehour_return + 1) * K2 ^ 0.5
+    # sharerate = lastonehour_share / (lastonehour_play + 1000)
+    # backrate = lastonehour_return / (lastonehour_share + 10)
+    # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+
+    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)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    df['score'] = df['share_rate'] ** 0.5 * df['back_rate'] ** 2 * df['log_back'] * df['K2'] ** 0.5
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+def cal_score_with_back_rate_by_rank_weighting(df, param):
+    """
+    add by sunmingze 20231123
+    计算score
+    :param df: 特征数据
+    :param param: 规则参数
+    :return:
+    """
+    # score计算公式: score =  1 / sharerate(rank)^0.5 + 5 / backrate(rank)^0.5 + 10 / LOG(lastonehour_return +1)(rank) ^0.5
+    #   +  1 / K2(rank)^0.5
+    # sharerate = lastonehour_share / (lastonehour_play + 1000)
+    # backrate = lastonehour_return / (lastonehour_share + 10)
+    # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+
+    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)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
+    elif param.get('view_type', None) == 'video-show-region':
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
+    else:
+        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+
+    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
+
+    # 分别的得到sharerate、backrate、K值、return人数的序关系
+    df['rank_by_sharerate'] = df['share_rate'].rank(ascending=0, method='dense')
+    df['rank_by_backrate'] = df['back_rate'].rank(ascending=0, method='dense')
+    df['rank_by_K2'] = df['K2'].rank(ascending=0, method='dense')
+    df['rank_by_logback'] = df['log_back'].rank(ascending=0, method='dense')
+
+    # 计算基于序的加法关系函数
+    df['score'] = 1/(df['rank_by_sharerate'] + 10) + 5/(df['rank_by_backrate'] + 10)
+    df['score'] = df['score'] + 5/(df['rank_by_logback'] + 10) + 1/(df['rank_by_K2'] + 10)
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+
+
+def cal_score(df, param):
+    if param.get('return_data', None) == 'share_region_return':
+        if param.get('score_func', None) == 'multiply_return_retention':
+            df = cal_score_multiply_return_retention_with_new_return(df=df, param=param)
+        else:
+            df = cal_score_with_new_return(df=df, param=param)
+    else:
+        if param.get('score_func', None) == 'add_backrate*log(return+1)':
+            df = cal_score_add_return(df=df, param=param)
+        elif param.get('score_func', None) == 'multiply_return_retention':
+            df = cal_score_multiply_return_retention(df=df, param=param)
+        elif param.get('score_func', None) == 'update_backrate':
+            df = cal_score_update_backrate(df=df, param=param)
+        elif param.get('score_func', None) == 'back_view0':
+            df = cal_score_with_back_view0(df=df, param=param)
+        elif param.get('score_func', None) == 'back_view1':
+            df = cal_score_with_back_view1(df=df, param=param)
+        elif param.get('score_func', None) == 'back_rate_exponential_weighting1':
+            df = cal_score_with_back_rate_exponential_weighting1(df=df, param=param)
+        elif param.get('score_func', None) == 'back_rate_exponential_weighting2':
+            df = cal_score_with_back_rate_exponential_weighting2(df=df, param=param)
+        elif param.get('score_func', None) == 'back_rate_rank_weighting':
+            df = cal_score_with_back_rate_by_rank_weighting(df=df, param=param)
+        else:
+            df = cal_score_initial(df=df, param=param)
+    return df
+
+
+def add_func1(initial_df, pre_h_df):
+    """当前小时级数据与前几个小时数据合并"""
+    score_list = initial_df['score'].to_list()
+    if len(score_list) > 0:
+        min_score = min(score_list)
+    else:
+        min_score = 0
+    pre_h_df = pre_h_df[pre_h_df['score'] > min_score]
+    df = pd.concat([initial_df, pre_h_df], ignore_index=True)
+    # videoid去重，保留分值高
+    df['videoid'] = df['videoid'].astype(int)
+    df = df.sort_values(by=['score'], ascending=False)
+    df = df.drop_duplicates(subset=['videoid'], keep="first")
+    return df
+
+
+def add_func2(initial_df, pre_h_df):
+    """当前小时级数据与前几个小时数据合并: 当前小时存在的视频以当前小时为准，否则以高分为主"""
+    score_list = initial_df['score'].to_list()
+    if len(score_list) > 0:
+        min_score = min(score_list)
+    else:
+        min_score = 0
+    initial_video_id_list = initial_df['videoid'].to_list()
+    pre_h_df = pre_h_df[pre_h_df['score'] > min_score]
+    pre_h_df = pre_h_df[~pre_h_df['videoid'].isin(initial_video_id_list)]
+
+    df = pd.concat([initial_df, pre_h_df], ignore_index=True)
+    # videoid去重，保留分值高
+    df['videoid'] = df['videoid'].astype(int)
+    df = df.sort_values(by=['score'], ascending=False)
+    df = df.drop_duplicates(subset=['videoid'], keep="first")
+    return df
+
+
+def add_videos(initial_df, now_date, rule_key, region, data_key, hour_count, top, add_func):
+    """
+    地域小时级数据列表中增加前6h优质视频
+    :param initial_df: 地域小时级筛选结果
+    :param now_date:
+    :param data_key:
+    :param region:
+    :param rule_key:
+    :param hour_count: 前几个小时, type-int
+    :param top: type-int
+    :return: df
+    """
+    redis_helper = RedisHelper()
+    pre_h_data = []
+    for i in range(1, hour_count+1):
+        pre_date = now_date - datetime.timedelta(hours=i)
+        pre_h = pre_date.hour
+        pre_h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \
+                                f"{datetime.datetime.strftime(pre_date, '%Y%m%d')}:{pre_h}"
+        pre_h_top_data = redis_helper.get_data_zset_with_index(key_name=pre_h_recall_key_name,
+                                                               start=0, end=top-1,
+                                                               desc=True, with_scores=True)
+        if pre_h_top_data is None:
+            continue
+        pre_h_data.extend(pre_h_top_data)
+    pre_h_df = pd.DataFrame(data=pre_h_data, columns=['videoid', 'score'])
+    if add_func == 'func2':
+        df = add_func2(initial_df=initial_df, pre_h_df=pre_h_df)
+    else:
+        df = add_func1(initial_df=initial_df, pre_h_df=pre_h_df)
+    return df
+
+
+def video_rank(df, now_date, now_h, rule_key, param, region, data_key, rule_rank_h_flag,
+               add_videos_with_pre_h=False, hour_count=0):
+    """
+    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
+    :param df:
+    :param now_date:
+    :param now_h:
+    :param rule_key: 小时级数据进入条件
+    :param param: 小时级数据进入条件参数
+    :param region: 所属地域
+    :return:
+    """
+    redis_helper = RedisHelper()
+
+    # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
+    return_count = param.get('return_count', 1)
+    score_value = param.get('score_rule', 0)
+    platform_return_rate = param.get('platform_return_rate', 0)
+    h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= score_value)
+                     & (df['platform_return_rate'] >= platform_return_rate)]
+
+    # videoid重复时，保留分值高
+    h_recall_df = h_recall_df.sort_values(by=['score'], ascending=False)
+    h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first')
+    h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
+
+    # 增加打捞的优质视频
+    if add_videos_with_pre_h is True:
+        add_func = param.get('add_func', None)
+        h_recall_df = add_videos(initial_df=h_recall_df, now_date=now_date, rule_key=rule_key,
+                                 region=region, data_key=data_key, hour_count=hour_count, top=10, add_func=add_func)
+
+    h_recall_videos = h_recall_df['videoid'].to_list()
+    # log_.info(f'h_recall videos count = {len(h_recall_videos)}')
+
+    # 视频状态过滤
+    if data_key in ['data7', ]:
+        filtered_videos = filter_video_status_app(h_recall_videos)
+    else:
+        filtered_videos = filter_video_status(h_recall_videos)
+    # log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
+
+    # 屏蔽视频过滤
+    shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
+    shield_key_name_list = shield_config.get(region, None)
+    if shield_key_name_list is not None:
+        filtered_videos = filter_shield_video(video_ids=filtered_videos, shield_key_name_list=shield_key_name_list)
+        # log_.info(f"shield filtered_videos count = {len(filtered_videos)}")
+
+    # 涉政视频过滤
+    political_filter = param.get('political_filter', None)
+    if political_filter is True:
+        # log_.info(f"political filter videos count = {len(filtered_videos)}")
+        filtered_videos = filter_political_videos(video_ids=filtered_videos)
+        # log_.info(f"political filtered videos count = {len(filtered_videos)}")
+
+    # 写入对应的redis
+    h_video_ids = []
+    by_30day_rule_key = param.get('30day_rule_key', None)
+    if by_30day_rule_key is not None:
+        # 与相对30天列表去重
+        h_video_ids = get_day_30day_videos(now_date=now_date, data_key=data_key, rule_key=by_30day_rule_key)
+        # log_.info(f"h_video_ids count = {len(h_video_ids)}")
+        if h_video_ids is not None:
+            filtered_videos = [video_id for video_id in filtered_videos if int(video_id) not in h_video_ids]
+            # log_.info(f"filtered_videos count = {len(filtered_videos)}")
+
+    h_recall_result = {}
+    for video_id in filtered_videos:
+        score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
+        # print(score)
+        h_recall_result[int(video_id)] = float(score)
+        h_video_ids.append(int(video_id))
+    h_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \
+        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    if len(h_recall_result) > 0:
+        # log_.info(f"h_recall_result count = {len(h_recall_result)}")
+        redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=2 * 24 * 3600)
+        # 限流视频score调整
+        update_limit_video_score(initial_videos=h_recall_result, key_name=h_recall_key_name)
+        # 清空线上过滤应用列表
+        # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}")
+
+    h_rule_key = param.get('h_rule_key', None)
+    region_24h_rule_key = param.get('region_24h_rule_key', 'rule1')
+    by_24h_rule_key = param.get('24h_rule_key', None)
+    by_48h_rule_key = param.get('48h_rule_key', None)
+    dup_remove = param.get('dup_remove', True)
+    # 与其他召回视频池去重，存入对应的redis
+    dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, h_rule_key=h_rule_key,
+                 region_24h_rule_key=region_24h_rule_key, by_24h_rule_key=by_24h_rule_key,
+                 by_48h_rule_key=by_48h_rule_key, region=region, data_key=data_key,
+                 rule_rank_h_flag=rule_rank_h_flag, political_filter=political_filter,
+                 shield_config=shield_config, dup_remove=dup_remove)
+
+
+def dup_data(h_video_ids, initial_key_name, dup_key_name, region, political_filter, shield_config, dup_remove):
+    redis_helper = RedisHelper()
+    if redis_helper.key_exists(key_name=initial_key_name):
+        initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True)
+        # 屏蔽视频过滤
+        initial_video_ids = [int(video_id) for video_id, _ in initial_data]
+        shield_key_name_list = shield_config.get(region, None)
+        if shield_key_name_list is not None:
+            initial_video_ids = filter_shield_video(video_ids=initial_video_ids,
+                                                    shield_key_name_list=shield_key_name_list)
+        # 涉政视频过滤
+        if political_filter is True:
+            initial_video_ids = filter_political_videos(video_ids=initial_video_ids)
+
+        dup_data = {}
+        # 视频去重逻辑
+        if dup_remove is True:
+            for video_id, score in initial_data:
+                if int(video_id) not in h_video_ids and int(video_id) in initial_video_ids:
+                    dup_data[int(video_id)] = score
+                    h_video_ids.append(int(video_id))
+        else:
+            for video_id, score in initial_data:
+                if int(video_id) in initial_video_ids:
+                    dup_data[int(video_id)] = score
+
+        if len(dup_data) > 0:
+            redis_helper.add_data_with_zset(key_name=dup_key_name, data=dup_data, expire_time=2 * 24 * 3600)
+            # 限流视频score调整
+            update_limit_video_score(initial_videos=dup_data, key_name=dup_key_name)
+    return h_video_ids
+
+
+def dup_to_redis(h_video_ids, now_date, now_h, rule_key, h_rule_key, region_24h_rule_key, by_24h_rule_key, by_48h_rule_key,
+                 region, data_key, rule_rank_h_flag, political_filter, shield_config, dup_remove):
+    """将地域分组小时级数据与其他召回视频池去重，存入对应的redis"""
+    # ##### 去重更新不区分地域小时级列表，并另存为redis中
+    if h_rule_key is not None:
+        h_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_BY_H_H}{data_key}:{h_rule_key}:" \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_dup_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H_H}{region}:{data_key}:{rule_key}:" \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_key_name,
+                               dup_key_name=h_dup_key_name, region=region, political_filter=political_filter,
+                               shield_config=shield_config, dup_remove=dup_remove)
+
+    # ##### 去重更新地域分组小时级24h列表，并另存为redis中
+    region_24h_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}:{data_key}:{region_24h_rule_key}:" \
+        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    region_24h_dup_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{data_key}:{rule_key}:" \
+        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=region_24h_key_name,
+                           dup_key_name=region_24h_dup_key_name, region=region, political_filter=political_filter,
+                           shield_config=shield_config, dup_remove=dup_remove)
+
+    if rule_rank_h_flag == '48h':
+
+        # ##### 去重小程序相对48h更新结果，并另存为redis中
+        h_48h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_48H}{data_key}:{by_48h_rule_key}:" \
+                         f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_48h_dup_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_48H_H}{region}:{data_key}:{rule_key}:" \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_48h_key_name,
+                               dup_key_name=h_48h_dup_key_name, region=region, political_filter=political_filter,
+                               shield_config=shield_config, dup_remove=dup_remove)
+
+        # ##### 去重小程序相对48h 筛选后剩余数据 更新结果，并另存为redis中
+        if by_48h_rule_key == 'rule1':
+            other_h_48h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_48H_OTHER}{data_key}:" \
+                                   f"{by_48h_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+            other_h_48h_dup_key_name = \
+                f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_48H_H}{region}:{data_key}:{rule_key}:" \
+                f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+            h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=other_h_48h_key_name,
+                                   dup_key_name=other_h_48h_dup_key_name, region=region,
+                                   political_filter=political_filter, shield_config=shield_config,
+                                   dup_remove=dup_remove)
+
+    else:
+        # ##### 去重小程序相对24h更新结果，并另存为redis中
+        h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{by_24h_rule_key}:" \
+                         f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_24h_dup_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{data_key}:{rule_key}:" \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_24h_key_name,
+                               dup_key_name=h_24h_dup_key_name, region=region, political_filter=political_filter,
+                               shield_config=shield_config, dup_remove=dup_remove)
+
+        # ##### 去重小程序相对24h 筛选后剩余数据 更新结果，并另存为redis中
+        # if by_24h_rule_key in ['rule3', 'rule4']:
+        other_h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:" \
+                               f"{by_24h_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        other_h_24h_dup_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{data_key}:{rule_key}:" \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=other_h_24h_key_name,
+                               dup_key_name=other_h_24h_dup_key_name, region=region, political_filter=political_filter,
+                               shield_config=shield_config, dup_remove=dup_remove)
+
+    # ##### 去重小程序模型更新结果，并另存为redis中
+    # model_key_name = get_rov_redis_key(now_date=now_date)
+    # model_data_dup_key_name = \
+    #     f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region}:{data_key}:{rule_key}:" \
+    #     f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    # h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=model_key_name,
+    #                        dup_key_name=model_data_dup_key_name, region=region)
+
+
+def merge_df(df_left, df_right):
+    """
+    df按照videoid, code 合并，对应特征求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid', 'code']
+    for feature in features:
+        if feature in ['apptype', 'videoid', 'code']:
+            continue
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+        feature_list.append(feature)
+    return df_merged[feature_list]
+
+
+def merge_df_with_score(df_left, df_right):
+    """
+    df 按照[videoid, code]合并，平台回流人数、回流人数、分数 分别求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid', 'code', 'lastonehour_return', 'platform_return', 'score']
+    for feature in feature_list[2:]:
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+    return df_merged[feature_list]
+
+
+def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
+                        rule_rank_h_flag, add_videos_with_pre_h, hour_count):
+    log_.info(f"region = {region} start...")
+    # 计算score
+    region_df = df_merged[df_merged['code'] == region]
+    log_.info(f'region = {region}, region_df count = {len(region_df)}')
+    score_df = cal_score(df=region_df, param=rule_param)
+    video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=rule_key, param=rule_param,
+               region=region, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag,
+               add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count)
+    log_.info(f"region = {region} end!")
+
+
+def process_with_region2(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
+                         rule_rank_h_flag, add_videos_with_pre_h, hour_count):
+    log_.info(f"region = {region} start...")
+    region_score_df = df_merged[df_merged['code'] == region]
+    log_.info(f'region = {region}, region_score_df count = {len(region_score_df)}')
+    video_rank(df=region_score_df, now_date=now_date, now_h=now_h, region=region,
+               rule_key=rule_key, param=rule_param, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag,
+               add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count)
+    log_.info(f"region = {region} end!")
+
+
+def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag):
+    log_.info(f"app_type = {app_type} start...")
+    data_params_item = params.get('data_params')
+    rule_params_item = params.get('rule_params')
+    task_list = []
+    for param in params.get('params_list'):
+        data_key = param.get('data')
+        data_param = data_params_item.get(data_key)
+        log_.info(f"data_key = {data_key}, data_param = {data_param}")
+        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+        df_merged = reduce(merge_df, df_list)
+
+        rule_key = param.get('rule')
+        rule_param = rule_params_item.get(rule_key)
+        log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+        task_list.extend(
+            [
+                gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
+                             now_date, now_h, rule_rank_h_flag)
+                for region in region_code_list
+            ]
+        )
+    gevent.joinall(task_list)
+    log_.info(f"app_type = {app_type} end!")
+
+
+    # log_.info(f"app_type = {app_type}")
+    # task_list = []
+    # for data_key, data_param in params['data_params'].items():
+    #     log_.info(f"data_key = {data_key}, data_param = {data_param}")
+    #     df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+    #     df_merged = reduce(merge_df, df_list)
+    #     for rule_key, rule_param in params['rule_params'].items():
+    #         log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+    #         task_list.extend(
+    #             [
+    #                 gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
+    #                              now_date, now_h)
+    #                 for region in region_code_list
+    #             ]
+    #         )
+    # gevent.joinall(task_list)
+
+
+def copy_data_for_city(region, city_code, data_key, rule_key, now_date, now_h, shield_config):
+    """copy 对应数据到城市对应redis，并做相应屏蔽视频过滤"""
+    log_.info(f"city_code = {city_code} start ...")
+    redis_helper = RedisHelper()
+    key_prefix_list = [
+        config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,  # 地域小时级
+        config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H,  # 地域相对24h
+        config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H,  # 不区分地域相对24h
+        config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H,  # 不区分地域相对24h筛选后
+        config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H,  # rov大列表
+    ]
+    for key_prefix in key_prefix_list:
+        region_key = f"{key_prefix}{region}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        city_key = f"{key_prefix}{city_code}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        if not redis_helper.key_exists(key_name=region_key):
+            continue
+        region_data = redis_helper.get_all_data_from_zset(key_name=region_key, with_scores=True)
+        if not region_data:
+            continue
+        # 屏蔽视频过滤
+        region_video_ids = [int(video_id) for video_id, _ in region_data]
+        shield_key_name_list = shield_config.get(city_code, None)
+        # shield_key_name_list = config_.SHIELD_CONFIG.get(city_code, None)
+        if shield_key_name_list is not None:
+            filtered_video_ids = filter_shield_video(video_ids=region_video_ids,
+                                                     shield_key_name_list=shield_key_name_list)
+        else:
+            filtered_video_ids = region_video_ids
+        city_data = {}
+        for video_id, score in region_data:
+            if int(video_id) in filtered_video_ids:
+                city_data[int(video_id)] = score
+
+        if len(city_data) > 0:
+            redis_helper.add_data_with_zset(key_name=city_key, data=city_data, expire_time=2 * 24 * 3600)
+
+    log_.info(f"city_code = {city_code} end!")
+
+
+def process_with_param(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag):
+    log_.info(f"param = {param} start...")
+
+    data_key = param.get('data')
+    data_param = data_params_item.get(data_key)
+    log_.info(f"data_key = {data_key}, data_param = {data_param}")
+    rule_key = param.get('rule')
+    rule_param = rule_params_item.get(rule_key)
+    log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+    merge_func = rule_param.get('merge_func', None)
+    # 是否在地域小时级数据中增加打捞的优质视频
+    add_videos_with_pre_h = rule_param.get('add_videos_with_pre_h', False)
+    hour_count = rule_param.get('hour_count', 0)
+
+    if merge_func == 2:
+        score_df_list = []
+        for apptype, weight in data_param.items():
+            df = feature_df[feature_df['apptype'] == apptype]
+            # 计算score
+            score_df = cal_score(df=df, param=rule_param)
+            score_df['score'] = score_df['score'] * weight
+            score_df_list.append(score_df)
+        # 分数合并
+        df_merged = reduce(merge_df_with_score, score_df_list)
+        # 更新平台回流比
+        df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['lastonehour_return']
+        task_list = [
+            gevent.spawn(process_with_region2,
+                         region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag,
+                         add_videos_with_pre_h, hour_count)
+            for region in region_code_list
+        ]
+    else:
+        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+        df_merged = reduce(merge_df, df_list)
+        task_list = [
+            gevent.spawn(process_with_region,
+                         region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag,
+                         add_videos_with_pre_h, hour_count)
+            for region in region_code_list
+        ]
+
+    gevent.joinall(task_list)
+
+    # 特殊城市视频数据准备
+    # 屏蔽视频过滤
+    shield_config = rule_param.get('shield_config', config_.SHIELD_CONFIG)
+    for region, city_list in config_.REGION_CITY_MAPPING.items():
+        t = [
+            gevent.spawn(
+                copy_data_for_city,
+                region, city_code, data_key, rule_key, now_date, now_h, shield_config
+            )
+            for city_code in city_list
+        ]
+        gevent.joinall(t)
+
+    log_.info(f"param = {param} end!")
+
+
+def rank_by_h(project, table, now_date, now_h, rule_params, region_code_list, rule_rank_h_flag):
+    # 获取特征数据
+    feature_df = get_feature_data(project=project, table=table, now_date=now_date)
+    feature_df['apptype'] = feature_df['apptype'].astype(int)
+    data_params_item = rule_params.get('data_params')
+    rule_params_item = rule_params.get('rule_params')
+    params_list = rule_params.get('params_list')
+    pool = multiprocessing.Pool(processes=len(params_list))
+    for param in params_list:
+        pool.apply_async(
+            func=process_with_param,
+            args=(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag)
+        )
+    pool.close()
+    pool.join()
+
+
+
+
+    # pool = multiprocessing.Pool(processes=len(config_.APP_TYPE))
+    # for app_type, params in rule_params.items():
+    #     pool.apply_async(func=process_with_app_type,
+    #                      args=(app_type, params, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag))
+    # pool.close()
+    # pool.join()
+
+    """
+    for app_type, params in rule_params.items():
+        log_.info(f"app_type = {app_type} start...")
+        data_params_item = params.get('data_params')
+        rule_params_item = params.get('rule_params')
+        for param in params.get('params_list'):
+            log_.info(f"param = {param} start...")
+            data_key = param.get('data')
+            data_param = data_params_item.get(data_key)
+            log_.info(f"data_key = {data_key}, data_param = {data_param}")
+            df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+            df_merged = reduce(merge_df, df_list)
+            rule_key = param.get('rule')
+            rule_param = rule_params_item.get(rule_key)
+            log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+
+            task_list = []
+            for region in region_code_list:
+                t = Thread(target=process_with_region,
+                           args=(region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h)
+                           )
+                t.start()
+                task_list.append(t)
+            for t in task_list:
+                t.join()
+            log_.info(f"param = {param} end!")
+        log_.info(f"app_type = {app_type} end!")
+    """
+
+    # for app_type, params in rule_params.items():
+    #     log_.info(f"app_type = {app_type}")
+    #     for data_key, data_param in params['data_params'].items():
+    #         log_.info(f"data_key = {data_key}, data_param = {data_param}")
+    #         df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+    #         df_merged = reduce(merge_df, df_list)
+    #         for rule_key, rule_param in params['rule_params'].items():
+    #             log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+    #             task_list = [
+    #                 gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h)
+    #                 for region in region_code_list
+    #             ]
+    #             gevent.joinall(task_list)
+
+    # rank
+    # for key, value in rule_params.items():
+    #     log_.info(f"rule = {key}, param = {value}")
+    #     for region in region_code_list:
+    #         log_.info(f"region = {region}")
+    #         # 计算score
+    #         region_df = feature_df[feature_df['code'] == region]
+    #         log_.info(f'region_df count = {len(region_df)}')
+    #         score_df = cal_score(df=region_df, param=value)
+    #         video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value, region=region)
+    #         # to-csv
+    #         score_filename = f"score_{region}_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
+    #         score_df.to_csv(f'./data/{score_filename}')
+    #         # to-logs
+    #         log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d%H'),
+    #                    "region_code": region,
+    #                    "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,
+    #                    "rule_key": key,
+    #                    # "score_df": score_df[['videoid', 'score']]
+    #                    }
+    #                   )
+
+
+def h_bottom_process(param, rule_params_item, region_code_list, key_prefix, redis_dt, redis_h,
+                     now_date, now_h, rule_rank_h_flag):
+    redis_helper = RedisHelper()
+    data_key = param.get('data')
+    rule_key = param.get('rule')
+    rule_param = rule_params_item.get(rule_key)
+    log_.info(f"data_key = {data_key}, rule_key = {rule_key}, rule_param = {rule_param}")
+    h_rule_key = rule_param.get('h_rule_key', None)
+    region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1')
+    by_24h_rule_key = rule_param.get('24h_rule_key', None)
+    by_48h_rule_key = rule_param.get('48h_rule_key', None)
+    # 涉政视频过滤
+    political_filter = param.get('political_filter', None)
+    # 屏蔽视频过滤
+    shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
+    dup_remove = param.get('dup_remove', True)
+    for region in region_code_list:
+        log_.info(f"region = {region}")
+        key_name = f"{key_prefix}{region}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+        initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+        if initial_data is None:
+            initial_data = []
+        final_data = dict()
+        h_video_ids = []
+        for video_id, score in initial_data:
+            final_data[video_id] = score
+            h_video_ids.append(int(video_id))
+        # 存入对应的redis
+        final_key_name = \
+            f"{key_prefix}{region}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+        if len(final_data) > 0:
+            redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 24 * 3600)
+        # 与其他召回视频池去重，存入对应的redis
+        dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, h_rule_key=h_rule_key,
+                     region_24h_rule_key=region_24h_rule_key, region=region,
+                     data_key=data_key, by_24h_rule_key=by_24h_rule_key,
+                     by_48h_rule_key=by_48h_rule_key, rule_rank_h_flag=rule_rank_h_flag,
+                     political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove)
+    # 特殊城市视频数据准备
+    for region, city_list in config_.REGION_CITY_MAPPING.items():
+        t = [
+            gevent.spawn(
+                copy_data_for_city,
+                region, city_code, data_key, rule_key, now_date, now_h, shield_config
+            )
+            for city_code in city_list
+        ]
+        gevent.joinall(t)
+
+
+def h_rank_bottom(now_date, now_h, rule_params, region_code_list, rule_rank_h_flag):
+    """未按时更新数据，用上一小时结果作为当前小时的数据"""
+    # 获取rov模型结果
+    # redis_helper = RedisHelper()
+    if now_h == 0:
+        redis_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
+        redis_h = 23
+    else:
+        redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+        redis_h = now_h - 1
+
+    # 以上一小时的地域分组数据作为当前小时的数据
+    key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
+    rule_params_item = rule_params.get('rule_params')
+    params_list = rule_params.get('params_list')
+    pool = multiprocessing.Pool(processes=len(params_list))
+    for param in params_list:
+        pool.apply_async(
+            func=h_bottom_process,
+            args=(param, rule_params_item, region_code_list, key_prefix, redis_dt, redis_h, now_date, now_h, rule_rank_h_flag)
+        )
+    pool.close()
+    pool.join()
+    # for param in rule_params.get('params_list'):
+    #     data_key = param.get('data')
+    #     rule_key = param.get('rule')
+    #     rule_param = rule_params_item.get(rule_key)
+    #     log_.info(f"data_key = {data_key}, rule_key = {rule_key}, rule_param = {rule_param}")
+    #     region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1')
+    #     by_24h_rule_key = rule_param.get('24h_rule_key', None)
+    #     by_48h_rule_key = rule_param.get('48h_rule_key', None)
+    #     # 涉政视频过滤
+    #     political_filter = param.get('political_filter', None)
+    #     # 屏蔽视频过滤
+    #     shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
+    #     for region in region_code_list:
+    #         log_.info(f"region = {region}")
+    #         key_name = f"{key_prefix}{region}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+    #         initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+    #         if initial_data is None:
+    #             initial_data = []
+    #         final_data = dict()
+    #         h_video_ids = []
+    #         for video_id, score in initial_data:
+    #             final_data[video_id] = score
+    #             h_video_ids.append(int(video_id))
+    #         # 存入对应的redis
+    #         final_key_name = \
+    #             f"{key_prefix}{region}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    #         if len(final_data) > 0:
+    #             redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 24 * 3600)
+    #         # 与其他召回视频池去重，存入对应的redis
+    #         dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key,
+    #                      region_24h_rule_key=region_24h_rule_key, region=region,
+    #                      data_key=data_key, by_24h_rule_key=by_24h_rule_key,
+    #                      by_48h_rule_key=by_48h_rule_key, rule_rank_h_flag=rule_rank_h_flag,
+    #                      political_filter=political_filter, shield_config=shield_config)
+    #     # 特殊城市视频数据准备
+    #     for region, city_list in config_.REGION_CITY_MAPPING.items():
+    #         t = [
+    #             gevent.spawn(
+    #                 copy_data_for_city,
+    #                 region, city_code, data_key, rule_key, now_date, now_h, shield_config
+    #             )
+    #             for city_code in city_list
+    #         ]
+    #         gevent.joinall(t)
+
+
+def h_timer_check():
+    try:
+        rule_rank_h_flag = sys.argv[1]
+        # rule_rank_h_flag = '24h'
+        if rule_rank_h_flag == '48h':
+            rule_params = config_.RULE_PARAMS_REGION_APP_TYPE_48H
+        else:
+            rule_params = RULE_PARAMS
+        project = config_.PROJECT_REGION_APP_TYPE
+        table = config_.TABLE_REGION_APP_TYPE
+        region_code_list = [code for region, code in region_code.items()]
+        now_date = datetime.datetime.today()
+        log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}, rule_rank_h_flag: {rule_rank_h_flag}")
+        now_h = datetime.datetime.now().hour
+        now_min = datetime.datetime.now().minute
+        if now_h == 0:
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list,
+                          rule_rank_h_flag=rule_rank_h_flag)
+            log_.info(f"region_h_data end!")
+            return
+        # 查看当前小时更新的数据是否已准备好
+        h_data_count = h_data_check(project=project, table=table, now_date=now_date)
+        if h_data_count > 0:
+            log_.info(f'region_h_data_count = {h_data_count}')
+            # 数据准备好，进行更新
+            rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params,
+                      project=project, table=table, region_code_list=region_code_list, rule_rank_h_flag=rule_rank_h_flag)
+            log_.info(f"region_h_data end!")
+        elif now_min > 40:
+            log_.info('h_recall data is None, use bottom data!')
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list,
+                          rule_rank_h_flag=rule_rank_h_flag)
+            log_.info(f"region_h_data end!")
+        else:
+            # 数据没准备好，1分钟后重新检查
+            Timer(60, h_timer_check).start()
+        # 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"region_h_data start...")
+    h_timer_check()

alg_recsys_recall_24h_noregion.py

@@ -0,0 +1,485 @@
+# -*- coding: utf-8 -*-
+import pandas as pd
+import traceback
+from functools import reduce
+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 utils import filter_video_status, check_table_partition_exits, filter_video_status_app, \
+    request_post, send_msg_to_feishu
+from config import set_config
+from log import Log
+
+config_, _ = set_config()
+log_ = Log()
+
+
+RULE_PARAMS = {
+    'rule_params': {
+        'rule66': {'cal_score_func': 2, 'return_count': 100, 'platform_return_rate': 0.001,
+                  'view_type': 'preview'},
+    },
+    'data_params': config_.DATA_PARAMS,
+    'params_list': [
+        {'data': 'data66', 'rule': 'rule66'},
+    ]
+}
+
+features = [
+    'apptype',
+    'videoid',
+    'preview人数',  # 过去24h预曝光人数
+    'view人数',  # 过去24h曝光人数
+    'play人数',  # 过去24h播放人数
+    'share人数',  # 过去24h分享人数
+    '回流人数',  # 过去24h分享，过去24h回流人数
+    'preview次数',  # 过去24h预曝光次数
+    'view次数',  # 过去24h曝光次数
+    'play次数',  # 过去24h播放次数
+    'share次数',  # 过去24h分享次数
+    'platform_return',
+    'platform_preview',
+    'platform_preview_total',
+    'platform_show',
+    'platform_show_total',
+    'platform_view',
+    'platform_view_total',
+]
+
+
+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, now_h):
+    """检查数据是否准备好"""
+    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:
+        # 23点开始到8点之前（不含8点），全部用22点生成那个列表
+        if now_h == 23:
+            dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
+        elif now_h < 8:
+            dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
+        else:
+            dt = datetime.strftime(now_date, '%Y%m%d%H')
+        check_res = check_table_partition_exits(date=dt, project=project, table=table)
+        if check_res:
+            sql = f'select * from {project}.{table} where dt = {dt}'
+            with odps.execute_sql(sql=sql).open_reader() as reader:
+                data_count = reader.count
+        else:
+            data_count = 0
+    except Exception as e:
+        data_count = 0
+    return data_count
+
+
+def get_feature_data(now_date, now_h, project, table):
+    """获取特征数据"""
+    # 23点开始到8点之前（不含8点），全部用22点生成那个列表
+    if now_h == 23:
+        dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
+    elif now_h < 8:
+        dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
+    else:
+        dt = datetime.strftime(now_date, '%Y%m%d%H')
+    log_.info({'feature_dt': dt})
+    # 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, param):
+    # score2计算公式: score = share次数/(view+1000)+0.01*return/(share次数+100)
+    df = df.fillna(0)
+    if param.get('view_type', None) == 'video-show':
+        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
+    elif param.get('view_type', None) == 'preview':
+        df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
+    else:
+        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['platform_return_rate'] = df['platform_return'] / df['回流人数']
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def cal_score(df, param):
+    # score计算公式: score1 = share次数/(view+1000)+0.01*return/(share次数+100)
+    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # score = 0.3 * score1 + 0.7 * K2
+    df = df.fillna(0)
+    if param.get('view_type', None) == 'video-show':
+        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
+        df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)
+    elif param.get('view_type', None) == 'preview':
+        df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
+        df['ctr'] = df['play人数'] / (df['preview人数'] + 1000)
+    else:
+        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
+        df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)
+
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+    df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
+    df['platform_return_rate'] = df['platform_return'] / df['回流人数']
+
+    df['score1'] = df['share_rate'] + 0.01 * df['back_rate']
+
+    click_score_rate = param.get('click_score_rate', None)
+    back_score_rate = param.get('click_score_rate', None)
+    if click_score_rate is not None:
+        df['score'] = (1 - click_score_rate) * df['score1'] + click_score_rate * df['K2']
+    elif back_score_rate is not None:
+        df['score'] = (1 - back_score_rate) * df['score1'] + back_score_rate * df['back_rate']
+    else:
+        df['score'] = df['score1']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def video_rank_h(df, now_date, now_h, rule_key, param, data_key, notify_backend):
+    """
+    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
+    :param df:
+    :param now_date:
+    :param now_h:
+    :param rule_key: 天级规则数据进入条件
+    :param param: 天级规则数据进入条件参数
+    :param data_key: 使用数据标识
+    :param notify_backend: 是否同步给后端标识
+    :return:
+    """
+    redis_helper = RedisHelper()
+    log_.info(f"videos_count = {len(df)}")
+
+    # videoid重复时，保留分值高
+    df = df.sort_values(by=['score'], ascending=False)
+    df = df.drop_duplicates(subset=['videoid'], keep='first')
+    df['videoid'] = df['videoid'].astype(int)
+
+    # 获取符合进入召回源条件的视频
+    return_count = param.get('return_count')
+    if return_count:
+        day_recall_df = df[df['回流人数'] > return_count]
+    else:
+        day_recall_df = df
+    platform_return_rate = param.get('platform_return_rate', 0)
+    day_recall_df = day_recall_df[day_recall_df['platform_return_rate'] > platform_return_rate]
+    day_recall_videos = day_recall_df['videoid'].to_list()
+    log_.info(f'h_by24h_recall videos count = {len(day_recall_videos)}')
+    # 视频状态过滤
+    if data_key in ['data7', ]:
+        filtered_videos = filter_video_status_app(day_recall_videos)
+    else:
+        filtered_videos = filter_video_status(day_recall_videos)
+    log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
+
+    # 写入对应的redis
+    now_dt = datetime.strftime(now_date, '%Y%m%d')
+    day_video_ids = []
+    day_recall_result = {}
+    # json_data = []
+    for video_id in filtered_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))
+        # json_data.append({'videoId': video_id, 'rovScore': float(score)})
+
+    h_24h_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{rule_key}:{now_dt}:{now_h}"
+    log_.info("h_24h_recall_key_name:redis:{}".format(h_24h_recall_key_name))
+    if len(day_recall_result) > 0:
+        log_.info(f"count = {len(day_recall_result)}, key = {h_24h_recall_key_name}")
+        redis_helper.add_data_with_zset(key_name=h_24h_recall_key_name, data=day_recall_result, expire_time=2 * 3600)
+        # 清空线上过滤应用列表
+        # redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")
+
+    # 去重筛选结果，保留剩余数据并写入Redis
+    all_videos = df['videoid'].to_list()
+    log_.info(f'h_by24h_recall all videos count = {len(all_videos)}')
+    # 视频状态过滤
+    if data_key in ['data7', ]:
+        all_filtered_videos = filter_video_status_app(all_videos)
+    else:
+        all_filtered_videos = filter_video_status(all_videos)
+    log_.info(f'all_filtered_videos count = {len(all_filtered_videos)}')
+    # 与筛选结果去重
+    other_videos = [video for video in all_filtered_videos if video not in day_video_ids]
+    log_.info(f'other_videos count = {len(other_videos)}')
+    # 写入对应的redis
+    other_24h_recall_result = {}
+    json_data = []
+    for video_id in other_videos:
+        score = df[df['videoid'] == video_id]['score']
+        other_24h_recall_result[int(video_id)] = float(score)
+        json_data.append({'videoId': video_id, 'rovScore': float(score)})
+    # other_h_24h_recall_key_name = \
+    #     f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{app_type}:{data_key}:{rule_key}:{now_dt}:{now_h}"
+    other_h_24h_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:{rule_key}:{now_dt}:{now_h}"
+    if len(other_24h_recall_result) > 0:
+        log_.info(f"count = {len(other_24h_recall_result)}")
+        redis_helper.add_data_with_zset(key_name=other_h_24h_recall_key_name, data=other_24h_recall_result,
+                                        expire_time=2 * 3600)
+    # 通知后端更新兜底视频数据
+    if notify_backend is True:
+        log_.info('json_data count = {}'.format(len(json_data[:5000])))
+        # log_.info(f"json_data = {json_data}")
+        result = request_post(request_url=config_.NOTIFY_BACKEND_updateFallBackVideoList_URL,
+                              request_data={'videos': json_data[:5000]})
+        if result is None:
+            log_.error('notify backend updateFallBackVideoList fail!')
+        elif result['code'] == 0:
+            log_.info('notify backend updateFallBackVideoList success!')
+        else:
+            log_.error('notify backend updateFallBackVideoList fail!')
+
+    # 去重更新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=23 * 3600)
+
+
+def merge_df(df_left, df_right):
+    """
+    df按照videoid 合并，对应特征求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid']
+    for feature in features:
+        if feature in ['apptype', 'videoid']:
+            continue
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+        feature_list.append(feature)
+    return df_merged[feature_list]
+
+
+def merge_df_with_score(df_left, df_right):
+    """
+    df 按照videoid合并，平台回流人数、回流人数、分数 分别求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid', '回流人数', 'platform_return', 'score']
+    for feature in feature_list[1:]:
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+    return df_merged[feature_list]
+
+
+def rank_by_h(now_date, now_h, rule_params, project, table):
+    # 获取特征数据
+    feature_df = get_feature_data(now_date=now_date, now_h=now_h, project=project, table=table)
+    feature_df['apptype'] = feature_df['apptype'].astype(int)
+    # rank
+    data_params_item = rule_params.get('data_params')
+    rule_params_item = rule_params.get('rule_params')
+    """
+    for param in rule_params.get('params_list'):
+        data_key = param.get('data')
+        data_param = data_params_item.get(data_key)
+        log_.info(f"data_key = {data_key}, data_param = {data_param}")
+        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+        df_merged = reduce(merge_df, df_list)
+
+        rule_key = param.get('rule')
+        rule_param = rule_params_item.get(rule_key)
+        log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+        # 计算score
+        cal_score_func = rule_param.get('cal_score_func', 1)
+        if cal_score_func == 2:
+            score_df = cal_score2(df=df_merged, param=rule_param)
+        else:
+            score_df = cal_score1(df=df_merged)
+        video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
+                     rule_key=rule_key, param=rule_param, data_key=data_key)
+    """
+
+    for param in rule_params.get('params_list'):
+        score_df_list = []
+        notify_backend = param.get('notify_backend', False)
+        data_key = param.get('data')
+        data_param = data_params_item.get(data_key)
+        log_.info(f"data_key = {data_key}, data_param = {data_param}")
+        rule_key = param.get('rule')
+        rule_param = rule_params_item.get(rule_key)
+        log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+        # cal_score_func = rule_param.get('cal_score_func', 1)
+        merge_func = rule_param.get('merge_func', 1)
+
+        if merge_func == 2:
+            for apptype, weight in data_param.items():
+                df = feature_df[feature_df['apptype'] == apptype]
+                # 计算score
+                score_df = cal_score(df=df, param=rule_param)
+                score_df['score'] = score_df['score'] * weight
+                score_df_list.append(score_df)
+            # 分数合并
+            df_merged = reduce(merge_df_with_score, score_df_list)
+            # 更新平台回流比
+            df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['回流人数']
+            video_rank_h(df=df_merged, now_date=now_date, now_h=now_h,
+                         rule_key=rule_key, param=rule_param, data_key=data_key,
+                         notify_backend=notify_backend)
+        else:
+            df_list = [feature_df[feature_df['apptype'] == apptype] for apptype, _ in data_param.items()]
+            df_merged = reduce(merge_df, df_list)
+            score_df = cal_score(df=df_merged, param=rule_param)
+            video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
+                         rule_key=rule_key, param=rule_param, data_key=data_key,
+                         notify_backend=notify_backend)
+
+    #     # 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_params):
+    """未按时更新数据，用模型召回数据作为当前的数据"""
+    redis_helper = RedisHelper()
+    if now_h == 0:
+        redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
+        redis_h = 23
+    else:
+        redis_dt = datetime.strftime(now_date, '%Y%m%d')
+        redis_h = now_h - 1
+    key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_24H, config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER]
+
+    for param in rule_params.get('params_list'):
+        data_key = param.get('data')
+        rule_key = param.get('rule')
+        log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
+        for key_prefix in key_prefix_list:
+            key_name = f"{key_prefix}{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+            initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+            if initial_data is None:
+                initial_data = []
+            final_data = dict()
+            for video_id, score in initial_data:
+                final_data[video_id] = score
+            # 存入对应的redis
+            final_key_name = \
+                f"{key_prefix}{data_key}:{rule_key}:{datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+            if len(final_data) > 0:
+                redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 3600)
+
+    """
+    for app_type, params in rule_params.items():
+        log_.info(f"app_type = {app_type}")
+        for param in params.get('params_list'):
+            data_key = param.get('data')
+            rule_key = param.get('rule')
+            log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
+            for key_prefix in key_prefix_list:
+                key_name = f"{key_prefix}{app_type}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+                initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+                if initial_data is None:
+                    initial_data = []
+                final_data = dict()
+                for video_id, score in initial_data:
+                    final_data[video_id] = score
+                # 存入对应的redis
+                final_key_name = \
+                    f"{key_prefix}{app_type}:{data_key}:{rule_key}:{datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+                if len(final_data) > 0:
+                    redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 3600)
+                # 清空线上过滤应用列表
+                # redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")
+    """
+
+def h_timer_check():
+    try:
+        project = config_.PROJECT_24H_APP_TYPE
+        table = config_.TABLE_24H_APP_TYPE
+        rule_params = RULE_PARAMS
+        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, now_h=now_h)
+        if now_h == 23 or now_h < 8:
+            log_.info(f'now_h = {now_h} use bottom data!')
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
+            log_.info(f"24h_data end!")
+        elif 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)
+            log_.info(f"24h_data end!")
+        elif now_min > 40:
+            log_.info('h_by24h_recall data is None, use bottom data!')
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
+            log_.info(f"24h_data end!")
+        else:
+            # 数据没准备好，1分钟后重新检查
+            Timer(60, h_timer_check).start()
+
+    except Exception as e:
+        log_.error(f"不区分地域24h数据更新失败, 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} - 不区分地域24h数据更新失败\n"
+                     f"exception: {e}\n"
+                     f"traceback: {traceback.format_exc()}"
+        )
+
+
+if __name__ == '__main__':
+    log_.info(f"24h_data start...")
+    h_timer_check()

alg_recsys_recall_24h_region.py

@@ -0,0 +1,524 @@
+# -*- coding: utf-8 -*-
+import multiprocessing
+import traceback
+import gevent
+import datetime
+import pandas as pd
+import math
+from functools import reduce
+from odps import ODPS
+from threading import Timer, Thread
+from utils import RedisHelper, get_data_from_odps, filter_video_status, check_table_partition_exits, \
+    filter_video_status_app, send_msg_to_feishu
+from config import set_config
+from log import Log
+
+# os.environ['NUMEXPR_MAX_THREADS'] = '16'
+
+config_, _ = set_config()
+log_ = Log()
+
+region_code = config_.REGION_CODE
+
+
+RULE_PARAMS = {
+    'rule_params': {
+        'rule66': {'view_type': 'video-show', 'return_count': 21, 'score_rule': 0,
+                  'platform_return_rate': 0.001},
+    },
+    'data_params': config_.DATA_PARAMS,
+    'params_list': [
+        {'data': 'data66', 'rule': 'rule66'},
+    ]
+}
+
+features = [
+    'apptype',
+    'code',  # 省份编码
+    'videoid',
+    'lastday_preview',  # 昨日预曝光人数
+    'lastday_view',  # 昨日曝光人数
+    'lastday_play',  # 昨日播放人数
+    'lastday_share',  # 昨日分享人数
+    'lastday_return',  # 昨日回流人数
+    'lastday_preview_total',  # 昨日预曝光次数
+    'lastday_view_total',  # 昨日曝光次数
+    'lastday_play_total',  # 昨日播放次数
+    'lastday_share_total',  # 昨日分享次数
+    'platform_return',
+    'platform_preview',
+    'platform_preview_total',
+    'platform_show',
+    'platform_show_total',
+    'platform_view',
+    'platform_view_total',
+]
+
+
+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 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')
+        check_res = check_table_partition_exits(date=dt, project=project, table=table)
+        if check_res:
+            sql = f'select * from {project}.{table} where dt = {dt}'
+            with odps.execute_sql(sql=sql).open_reader() as reader:
+                data_count = reader.count
+        else:
+            data_count = 0
+    except Exception as e:
+        data_count = 0
+    return data_count
+
+
+def get_feature_data(project, table, 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, param):
+    """
+    计算score
+    :param df: 特征数据
+    :param param:
+    :return:
+    """
+    # score计算公式: sharerate*backrate*logback*ctr
+    # sharerate = lastday_share/(lastday_play+1000)
+    # backrate = lastday_return/(lastday_share+10)
+    # ctr = lastday_play/(lastday_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
+    # score = sharerate * backrate * LOG(lastday_return+1) * K2
+
+    df = df.fillna(0)
+    df['share_rate'] = df['lastday_share'] / (df['lastday_play'] + 1000)
+    df['back_rate'] = df['lastday_return'] / (df['lastday_share'] + 10)
+    df['log_back'] = (df['lastday_return'] + 1).apply(math.log)
+    if param.get('view_type', None) == 'video-show':
+        df['ctr'] = df['lastday_play'] / (df['platform_show'] + 1000)
+    else:
+        df['ctr'] = df['lastday_play'] / (df['lastday_preview'] + 1000)
+    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
+    df['platform_return_rate'] = df['platform_return'] / df['lastday_return']
+
+    df['score1'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
+
+    click_score_rate = param.get('click_score_rate', None)
+    back_score_rate = param.get('click_score_rate', None)
+    if click_score_rate is not None:
+        df['score'] = (1 - click_score_rate) * df['score1'] + click_score_rate * df['K2']
+    elif back_score_rate is not None:
+        df['score'] = (1 - back_score_rate) * df['score1'] + back_score_rate * df['back_rate']
+    else:
+        df['score'] = df['score1']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+def video_rank(df, now_date, now_h, rule_key, param, region, data_key):
+    """
+    获取符合进入召回源条件的视频
+    :param df:
+    :param now_date:
+    :param now_h:
+    :param rule_key: 小时级数据进入条件
+    :param param: 小时级数据进入条件参数
+    :param region: 所属地域
+    :return:
+    """
+    redis_helper = RedisHelper()
+    # 获取符合进入召回源条件的视频
+    return_count = param.get('return_count', 1)
+    score_value = param.get('score_rule', 0)
+    platform_return_rate = param.get('platform_return_rate', 0)
+    h_recall_df = df[(df['lastday_return'] >= return_count) & (df['score'] >= score_value)
+                     & (df['platform_return_rate'] >= platform_return_rate)]
+    log_.info(f'h_recall_df count = {len(h_recall_df)}')
+    # videoid重复时，保留分值高
+    h_recall_df = h_recall_df.sort_values(by=['score'], ascending=False)
+    h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first')
+    h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
+    h_recall_videos = h_recall_df['videoid'].to_list()
+    log_.info(f'h_recall_videos count = {len(h_recall_videos)}')
+    log_.info('h_recall_videos:{}'.format('-'.join([str(i) for i in h_recall_videos])))
+
+    # 视频状态过滤
+    if data_key in ['data7', ]:
+        filtered_videos = filter_video_status_app(h_recall_videos)
+    else:
+        filtered_videos = filter_video_status(h_recall_videos)
+    log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
+
+    # 写入对应的redis
+    h_video_ids = []
+    day_recall_result = {}
+    for video_id in filtered_videos:
+        score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
+        # print(score)
+        day_recall_result[int(video_id)] = float(score)
+        h_video_ids.append(int(video_id))
+    day_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}:{data_key}:{rule_key}:" \
+        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+    log_.info("day_recall_result.type:{}".format(str(type(day_recall_result))))
+    log_.info("begin to write redis for day_recall_key_name:{} with {}".format(day_recall_key_name,
+                                                                               str(len(day_recall_result))))
+    if len(day_recall_result) > 0:
+        redis_helper.add_data_with_zset(key_name=day_recall_key_name, data=day_recall_result, expire_time=2 * 3600)
+        # 清空线上过滤应用列表
+        # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}")
+
+    # 与其他召回视频池去重，存入对应的redis
+    # dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, region=region)
+
+
+def merge_df(df_left, df_right):
+    """
+    df按照videoid, code 合并，对应特征求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid', 'code']
+    for feature in features:
+        if feature in ['apptype', 'videoid', 'code']:
+            continue
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+        feature_list.append(feature)
+    return df_merged[feature_list]
+
+
+def merge_df_with_score(df_left, df_right):
+    """
+    df 按照[videoid, code]合并，平台回流人数、回流人数、分数 分别求和
+    :param df_left:
+    :param df_right:
+    :return:
+    """
+    df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
+    df_merged.fillna(0, inplace=True)
+    feature_list = ['videoid', 'code', 'lastday_return', 'platform_return', 'score']
+    for feature in feature_list[2:]:
+        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
+    return df_merged[feature_list]
+
+
+def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_date, now_h):
+    log_.info(f"region = {region} start...")
+    # 计算score
+    region_df = df_merged[df_merged['code'] == region]
+    log_.info(f'region = {region}, region_df count = {len(region_df)}')
+    score_df = cal_score(df=region_df, param=rule_param)
+    video_rank(df=score_df, now_date=now_date, now_h=now_h, region=region,
+               rule_key=rule_key, param=rule_param, data_key=data_key)
+    log_.info(f"region = {region} end!")
+
+
+def process_with_region2(region, df_merged, data_key, rule_key, rule_param, now_date, now_h):
+    log_.info(f"region = {region} start...")
+    region_score_df = df_merged[df_merged['code'] == region]
+    log_.info(f'region = {region}, region_score_df count = {len(region_score_df)}')
+    video_rank(df=region_score_df, now_date=now_date, now_h=now_h, region=region,
+               rule_key=rule_key, param=rule_param, data_key=data_key)
+    log_.info(f"region = {region} end!")
+
+
+def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h):
+    log_.info(f"app_type = {app_type} start...")
+    data_params_item = params.get('data_params')
+    rule_params_item = params.get('rule_params')
+    for param in params.get('params_list'):
+        data_key = param.get('data')
+        data_param = data_params_item.get(data_key)
+        log_.info(f"data_key = {data_key}, data_param = {data_param}")
+        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+        df_merged = reduce(merge_df, df_list)
+
+        rule_key = param.get('rule')
+        rule_param = rule_params_item.get(rule_key)
+        log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+        task_list = [
+            gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
+                         now_date, now_h)
+            for region in region_code_list
+        ]
+        gevent.joinall(task_list)
+    log_.info(f"app_type = {app_type} end!")
+
+
+def process_with_param(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h):
+    log_.info(f"param = {param} start...")
+    data_key = param.get('data')
+    data_param = data_params_item.get(data_key)
+    log_.info(f"data_key = {data_key}, data_param = {data_param}")
+    rule_key = param.get('rule')
+    rule_param = rule_params_item.get(rule_key)
+    log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+    merge_func = rule_param.get('merge_func', None)
+    if merge_func == 2:
+        score_df_list = []
+        for apptype, weight in data_param.items():
+            df = feature_df[feature_df['apptype'] == apptype]
+            # 计算score
+            score_df = cal_score(df=df, param=rule_param)
+            score_df['score'] = score_df['score'] * weight
+            score_df_list.append(score_df)
+        # 分数合并
+        df_merged = reduce(merge_df_with_score, score_df_list)
+        # 更新平台回流比
+        df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['lastday_return']
+        task_list = [
+            gevent.spawn(process_with_region2, region, df_merged, data_key, rule_key, rule_param, now_date, now_h)
+            for region in region_code_list
+        ]
+    else:
+        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype, _ in data_param.items()]
+        df_merged = reduce(merge_df, df_list)
+        task_list = [
+            gevent.spawn(process_with_region, region, df_merged, data_key, rule_key, rule_param, now_date, now_h)
+            for region in region_code_list
+        ]
+
+    gevent.joinall(task_list)
+    log_.info(f"param = {param} end!")
+
+
+def rank_by_24h(project, table, now_date, now_h, rule_params, region_code_list):
+    # 获取特征数据
+    feature_df = get_feature_data(project=project, table=table, now_date=now_date)
+    feature_df['apptype'] = feature_df['apptype'].astype(int)
+    # rank
+    data_params_item = rule_params.get('data_params')
+    rule_params_item = rule_params.get('rule_params')
+    params_list = rule_params.get('params_list')
+    pool = multiprocessing.Pool(processes=len(params_list))
+    for param in params_list:
+        pool.apply_async(
+            func=process_with_param,
+            args=(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h)
+        )
+    pool.close()
+    pool.join()
+
+    """
+    pool = multiprocessing.Pool(processes=len(config_.APP_TYPE))
+    for app_type, params in rule_params.items():
+        pool.apply_async(func=process_with_app_type,
+                         args=(app_type, params, region_code_list, feature_df, now_date, now_h))
+    pool.close()
+    pool.join()
+    """
+
+    # for app_type, params in rule_params.items():
+    #     log_.info(f"app_type = {app_type}")
+    #     for data_key, data_param in params['data_params'].items():
+    #         log_.info(f"data_key = {data_key}, data_param = {data_param}")
+    #         df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
+    #         df_merged = reduce(merge_df, df_list)
+    #         for rule_key, rule_param in params['rule_params'].items():
+    #             log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
+    #             task_list = [
+    #                 gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
+    #                              now_date, now_h)
+    #                 for region in region_code_list
+    #             ]
+    #             gevent.joinall(task_list)
+
+
+    # for key, value in rule_params.items():
+    #     log_.info(f"rule = {key}, param = {value}")
+    #     for region in region_code_list:
+    #         log_.info(f"region = {region}")
+    #         # 计算score
+    #         region_df = feature_df[feature_df['code'] == region]
+    #         log_.info(f'region_df count = {len(region_df)}')
+    #         score_df = cal_score(df=region_df, param=value)
+    #         video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value, region=region)
+    #         # to-csv
+    #         score_filename = f"score_24h_{region}_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
+    #         score_df.to_csv(f'./data/{score_filename}')
+    #         # to-logs
+    #         log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d%H'),
+    #                    "region_code": region,
+    #                    "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H,
+    #                    "rule_key": key,
+    #                    # "score_df": score_df[['videoid', 'score']]
+    #                    })
+
+
+def dup_to_redis(h_video_ids, now_date, now_h, rule_key, region):
+    """将地域分组小时级数据与其他召回视频池去重，存入对应的redis"""
+    redis_helper = RedisHelper()
+
+    # ##### 去重小程序天级更新结果，并另存为redis中
+    day_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}rule2.{datetime.datetime.strftime(now_date, '%Y%m%d')}"
+    if redis_helper.key_exists(key_name=day_key_name):
+        day_data = redis_helper.get_all_data_from_zset(key_name=day_key_name, with_scores=True)
+        log_.info(f'day data count = {len(day_data)}')
+        day_dup = {}
+        for video_id, score in day_data:
+            if int(video_id) not in h_video_ids:
+                day_dup[int(video_id)] = score
+                h_video_ids.append(int(video_id))
+        log_.info(f"day data dup count = {len(day_dup)}")
+        day_dup_key_name = \
+            f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_DAY_24H}{region}.{rule_key}." \
+            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+        if len(day_dup) > 0:
+            redis_helper.add_data_with_zset(key_name=day_dup_key_name, data=day_dup, expire_time=23 * 3600)
+
+    # ##### 去重小程序模型更新结果，并另存为redis中
+    model_key_name = get_rov_redis_key(now_date=now_date)
+    model_data = redis_helper.get_all_data_from_zset(key_name=model_key_name, with_scores=True)
+    log_.info(f'model data count = {len(model_data)}')
+    model_data_dup = {}
+    for video_id, score in model_data:
+        if int(video_id) not in h_video_ids:
+            model_data_dup[int(video_id)] = score
+            h_video_ids.append(int(video_id))
+    log_.info(f"model data dup count = {len(model_data_dup)}")
+    model_data_dup_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_24H}{region}.{rule_key}." \
+        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    if len(model_data_dup) > 0:
+        redis_helper.add_data_with_zset(key_name=model_data_dup_key_name, data=model_data_dup, expire_time=23 * 3600)
+
+
+def h_rank_bottom(now_date, now_h, rule_params, region_code_list):
+    """未按时更新数据，用上一小时结果作为当前小时的数据"""
+    redis_helper = RedisHelper()
+    if now_h == 0:
+        redis_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
+        redis_h = 23
+    else:
+        redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+        redis_h = now_h - 1
+
+    # 以上一小时的地域分组数据作为当前小时的数据
+    key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H
+    for param in rule_params.get('params_list'):
+        data_key = param.get('data')
+        rule_key = param.get('rule')
+        log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
+        for region in region_code_list:
+            log_.info(f"region = {region}")
+            key_name = f"{key_prefix}{region}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+            initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+            if initial_data is None:
+                initial_data = []
+            final_data = dict()
+            h_video_ids = []
+            for video_id, score in initial_data:
+                final_data[video_id] = score
+                h_video_ids.append(int(video_id))
+            # 存入对应的redis
+            final_key_name = \
+                f"{key_prefix}{region}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+            if len(final_data) > 0:
+                redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 3600)
+
+    """
+    for app_type, params in rule_params.items():
+        log_.info(f"app_type = {app_type}")
+        for param in params.get('params_list'):
+            data_key = param.get('data')
+            rule_key = param.get('rule')
+            log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
+            for region in region_code_list:
+                log_.info(f"region = {region}")
+                key_name = f"{key_prefix}{region}:{app_type}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
+                initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
+                if initial_data is None:
+                    initial_data = []
+                final_data = dict()
+                h_video_ids = []
+                for video_id, score in initial_data:
+                    final_data[video_id] = score
+                    h_video_ids.append(int(video_id))
+                # 存入对应的redis
+                final_key_name = \
+                    f"{key_prefix}{region}:{app_type}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
+                if len(final_data) > 0:
+                    redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=2 * 3600)
+                # 清空线上过滤应用列表
+                # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}")
+
+                # 与其他召回视频池去重，存入对应的redis
+                # dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, region=region)
+    """
+
+
+def h_timer_check():
+    try:
+        rule_params = RULE_PARAMS
+        project = config_.PROJECT_REGION_24H_APP_TYPE
+        table = config_.TABLE_REGION_24H_APP_TYPE
+        region_code_list = [code for region, code in region_code.items() if code != '-1']
+        now_date = datetime.datetime.today()
+        now_h = datetime.datetime.now().hour
+        now_min = datetime.datetime.now().minute
+        log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
+        # 查看当天更新的数据是否已准备好
+        h_data_count = data_check(project=project, table=table, now_date=now_date)
+        if h_data_count > 0:
+            log_.info(f'region_24h_data_count = {h_data_count}')
+            # 数据准备好，进行更新
+            rank_by_24h(now_date=now_date, now_h=now_h, rule_params=rule_params,
+                        project=project, table=table, region_code_list=region_code_list)
+            log_.info(f"region_24h_data end!")
+        elif now_min > 40:
+            log_.info('24h_recall data is None, use bottom data!')
+            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list)
+            log_.info(f"region_24h_data end!")
+        else:
+            # 数据没准备好，1分钟后重新检查
+            Timer(60, h_timer_check).start()
+
+    except Exception as e:
+        log_.error(f"地域分组24h数据更新失败, 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} - 地域分组24h数据更新失败\n"
+                     f"exception: {e}\n"
+                     f"traceback: {traceback.format_exc()}"
+        )
+
+
+if __name__ == '__main__':
+    log_.info(f"region_24h_data start...")
+    h_timer_check()

alg_recsys_recall_hour_region_task.sh

@@ -0,0 +1,29 @@
+source /etc/profile
+echo $ROV_OFFLINE_ENV
+if [ ! -d "logs_dir" ]; then
+    # 如果文件夹不存在，则创建文件夹
+    mkdir logs_dir
+fi
+cur_time="`date +%Y%m%d`"
+cur_h="`date +%H`"
+echo "开始执行时间:{$cur_time}-{$cur_h}"
+
+if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
+  cd /root/zhangbo/rov-offline
+  /root/anaconda3/bin/python alg_recsys_recall_24h_noregion.py > "logs_dir/alg_recsys_recall_24h_noregion_{$cur_time}_{$cur_h}.log" &
+  /root/anaconda3/bin/python alg_recsys_recall_24h_region.py > "logs_dir/alg_recsys_recall_24h_region_{$cur_time}_{$cur_h}.log"
+  wait
+  echo "并行执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  /root/anaconda3/bin/python alg_recsys_recall_1h_region.py '24h' > "logs_dir/alg_recsys_recall_1h_region_{$cur_time}_{$cur_h}.log"
+  echo "结束执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  echo "all done"
+elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
+  cd /root/zhangbo/rov-offline
+  /root/anaconda3/bin/python alg_recsys_recall_24h_noregion.py > "logs_dir/alg_recsys_recall_24h_noregion_{$cur_time}_{$cur_h}.log" &
+  /root/anaconda3/bin/python alg_recsys_recall_24h_region.py > "logs_dir/alg_recsys_recall_24h_region_{$cur_time}_{$cur_h}.log"
+  wait
+  echo "并行执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  /root/anaconda3/bin/python alg_recsys_recall_1h_region.py '24h' > "logs_dir/alg_recsys_recall_1h_region_{$cur_time}_{$cur_h}.log"
+  echo "结束执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  echo "all done"
+fi

config.py

@@ -2550,18 +2550,21 @@ class TestConfig(BaseConfig):
 
     # 测试环境 过滤用mysql地址
     FILTER_MYSQL_INFO = {
-        'host': 'am-bp1g3ys9u00u483uc131930.ads.aliyuncs.com',
+        # 'host': 'am-bp1g3ys9u00u483uc131930.ads.aliyuncs.com',
-        'port': 3306,
-        'user': 'lv_manager',
-        'password': 'lv_manager@2020',
-        'db': 'longvideo',
-        'charset': 'utf8'
-        # 'host': 'am-bp15tqt957i3b3sgi131950.ads.aliyuncs.com',
         # 'port': 3306,
         # 'user': 'lv_manager',
         # 'password': 'lv_manager@2020',
         # 'db': 'longvideo',
         # 'charset': 'utf8'
+
+        ##### test环境的filter mysql会过滤掉所有数据，测试时先使用pro的filter mysql。 注意测试结束后切换注释。
+
+        'host': 'am-bp15tqt957i3b3sgi131950.ads.aliyuncs.com',
+        'port': 3306,
+        'user': 'lv_manager',
+        'password': 'lv_manager@2020',
+        'db': 'longvideo',
+        'charset': 'utf8'
     }
 
     # 日志服务配置