rov-offline/alg_recsys_recall_aftermerge.py

# -*- 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 my_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 my_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
        # 查看当前小时更新的数据是否已准备好
        if now_min < 45:
            log_.info('开始正常合并')
            # 数据准备好，进行更新
            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("合并5----------正常完成----------")
        else:
            log_.info('当前合并分钟超过45，预计执行无法完成，使用 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('----------当前分钟超过45，使用bottom的data，完成----------')
    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()