去重代码修正

alg_recsys_recall_aftermerge.py

@@ -0,0 +1,433 @@
+# -*- 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'
+        },
+        'rule67': {
+          'view_type': 'video-show-region', 'platform_return_rate': 0.001,
+        'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66', 'h_rule_key': 'rule66'
+         },
+         'rule68': {
+             'view_type': 'video-show-region', 'platform_return_rate': 0.001,
+             'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66',
+             'score_func': 'back_rate_exponential_weighting1'
+         },
+
+    },
+    'data_params': config_.DATA_PARAMS,
+    'params_list': [
+        # 532
+        {'data': 'data66', 'rule': 'rule66'},  # 523-> 523 & 518
+        {'data': 'data66', 'rule': 'rule67'},  # 523->510
+        {'data': 'data66', 'rule': 'rule68'},  # 523->514
+        # {'data': 'data66', 'rule': 'rule69'},  # 523->518
+    ],
+}
+
+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 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 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):
+
+    shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
+    political_filter = param.get('political_filter', None)
+    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}"
+    redis_helper = RedisHelper()
+    if redis_helper.key_exists(key_name=h_recall_key_name):
+        initial_data = redis_helper.get_all_data_from_zset(key_name=h_recall_key_name, with_scores=True)
+        h_video_ids = [int(video_id) for video_id, _ in initial_data]
+    else:
+        h_video_ids = []
+        log_.info("地域小时级别没有数据，下游不会过滤。")
+
+    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]
+
+        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"""
+
+    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}"
+        log_.info("开始去重【1小时 无地域，写入key的前缀是：{}".format(h_dup_key_name))
+        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}"
+    log_.info("开始去重【24小时 地域】，写入key的前缀是：{}".format(region_24h_dup_key_name))
+    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)
+
+
+    # ##### 去重小程序相对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}"
+    log_.info("开始去重【24小时 无地域】，写入key的前缀是：{}".format(region_24h_dup_key_name))
+    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中
+    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}"
+    log_.info("开始去重【24小时 无地域 other】，写入key的前缀是：{}".format(other_h_24h_dup_key_name))
+    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)
+
+
+
+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} 开始执行")
+    video_rank(df=None, 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} 完成执行")
+
+
+
+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):
+    data_key = param.get('data')
+    data_param = data_params_item.get(data_key)
+    rule_key = param.get('rule')
+    rule_param = rule_params_item.get(rule_key)
+    merge_func = rule_param.get('merge_func', None)
+    log_.info("数据采用:{},统计采用{}.".format(data_key, rule_key))
+    log_.info("具体的规则是:{}.".format(rule_param))
+    # 是否在地域小时级数据中增加打捞的优质视频
+    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:
+        pass
+    else:
+        task_list = [
+            gevent.spawn(process_with_region,
+                         region, None, 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)
+
+
+    log_.info(f"多进程的 param = {param} 完成执行!")
+
+
+def rank_by_h(project, table, now_date, now_h, rule_params, region_code_list, rule_rank_h_flag):
+    # 获取特征数据
+    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, None, now_date, now_h, rule_rank_h_flag)
+        )
+    pool.close()
+    pool.join()
+
+
+
+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()
+
+
+def h_timer_check():
+    try:
+        rule_rank_h_flag = "24h"
+        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"开始执行: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
+        now_h = datetime.datetime.now().hour
+        now_min = datetime.datetime.now().minute
+        if now_h == 0:
+            log_.info("当前时间{}小时，使用bottom的data，开始。".format(now_h))
+            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("----------当前时间{}小时，使用bottom的data，完成----------".format(now_h))
+            return
+        # 查看当前小时更新的数据是否已准备好
+        h_data_count = h_data_check(project=project, table=table, now_date=now_date)
+        if h_data_count > 0:
+            log_.info('上游数据表查询数据条数 h_data_count = {}，开始计算。'.format(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("----------正常完成----------")
+        elif now_min > 40:
+            log_.info('当前分钟超过40，预计执行无法完成，使用 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('----------当前分钟超过40，使用bottom的data，完成----------')
+        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_aftermerge.py:「去重」 开始执行")
+    h_timer_check()