同步 1小时

alg_recsys_recall_1h_noregion.py

@@ -0,0 +1,309 @@
+import pandas as pd
+import math
+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, send_msg_to_feishu
+from config import set_config
+from log import Log
+
+config_, _ = set_config()
+log_ = Log()
+
+RULE_PARAMS = {
+    'rule_params': {
+        'rule66': {'view_type': 'video-show', 'platform_return_rate': 0.001},
+    },
+    'data_params': config_.DATA_PARAMS,
+    'params_list': [
+        {'data': 'data66', 'rule': 'rule66'},
+    ],
+}
+
+features = [
+    'apptype',
+    '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',  # 不区分地域
+    '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 h_data_check(project, table, now_date):
+    """检查数据是否准备好"""
+    odps = ODPS(
+        access_id=config_.ODPS_CONFIG['ACCESSID'],
+        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
+        project=project,
+        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
+        connect_timeout=3000,
+        read_timeout=500000,
+        pool_maxsize=1000,
+        pool_connections=1000
+    )
+
+    try:
+        dt = datetime.strftime(now_date, '%Y%m%d%H')
+        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 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 = config_.RECALL_KEY_NAME_PREFIX_BY_H_H
+
+    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}")
+        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)
+
+
+def get_feature_data(project, table, now_date):
+    """获取特征数据"""
+    dt = datetime.strftime(now_date, '%Y%m%d%H')
+    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 = sharerate * backrate * 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)
+    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'] * df['log_back'] * df['K2']
+
+    df = df.sort_values(by=['score'], ascending=False)
+    return df
+
+
+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', 'lastonehour_return', '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 video_rank_h(df, now_date, now_h, rule_key, param, data_key):
+    """
+    获取符合进入召回源条件的视频
+    """
+    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)
+
+    # 获取符合进入召回源条件的视频
+    platform_return_rate = param.get('platform_return_rate', 0)
+    h_recall_df = df[df['platform_return_rate'] > platform_return_rate]
+    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)))
+
+    # 写入对应的redis
+    now_dt = datetime.strftime(now_date, '%Y%m%d')
+    h_video_ids = []
+    h_recall_result = {}
+    for video_id in filtered_videos:
+        score = h_recall_df[h_recall_df['videoid'] == video_id]['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_BY_H_H}{data_key}:{rule_key}:{now_dt}:{now_h}"
+
+    if len(h_recall_result) > 0:
+        log_.info(f"count = {len(h_recall_result)}, key = {h_recall_key_name}")
+        redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=2 * 3600)
+
+
+def rank_by_h(now_date, now_h, rule_params, project, table):
+    # 获取特征数据
+    feature_df = get_feature_data(now_date=now_date, project=project, table=table)
+    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'):
+        score_df_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}")
+        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', 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['lastonehour_return']
+            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)
+
+        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)
+
+
+def h_timer_check():
+    try:
+        project = config_.PROJECT_H_APP_TYPE
+        table = config_.TABLE_H_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
+
+        if now_h == 0:
+            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"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'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)
+            log_.info(f"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)
+            log_.info(f"h_data end!")
+        else:
+            # 数据没准备好，1分钟后重新检查
+            Timer(60, h_timer_check).start()
+
+    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("文件alg_recsys_recall_24h_noregion.py:「24小时无地域」 开始执行")
+    h_timer_check()