rov-server/recommend.py

import time
import multiprocessing
import traceback

from datetime import datetime

import config
from log import Log
from config import set_config
from video_recall import PoolRecall
from video_rank import video_rank, bottom_strategy, video_rank_by_w_h_rate
from db_helper import RedisHelper
import gevent
from utils import FilterVideos
import ast

log_ = Log()
config_ = set_config()


def video_position_recommend(mid, uid, app_type, videos):
    # videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
    #                          algo_type=algo_type, client_info=client_info)
    redis_helper = RedisHelper()
    pos1_vids = redis_helper.get_data_from_redis(config.BaseConfig.RECALL_POSITION1_KEY_NAME)
    pos2_vids = redis_helper.get_data_from_redis(config.BaseConfig.RECALL_POSITION2_KEY_NAME)
    if pos1_vids is not None:
        pos1_vids = ast.literal_eval(pos1_vids)
    if pos2_vids is not None:
        pos2_vids = ast.literal_eval(pos2_vids)

    pos1_vids = [] if pos1_vids is None else pos1_vids
    pos2_vids = [] if pos2_vids is None else pos2_vids

    pos1_vids = [int(i) for i in pos1_vids]
    pos2_vids = [int(i) for i in pos2_vids]

    filter_1 = FilterVideos(app_type=app_type, video_ids=pos1_vids, mid=mid, uid=uid)
    filter_2 = FilterVideos(app_type=app_type, video_ids=pos2_vids, mid=mid, uid=uid)
    t = [gevent.spawn(filter_1.filter_videos), gevent.spawn(filter_2.filter_videos)]
    gevent.joinall(t)
    filted_list = [i.get() for i in t]

    pos1_vids = filted_list[0]
    pos2_vids = filted_list[1]

    videos = positon_duplicate(pos1_vids, pos2_vids, videos)    

    if pos1_vids is not None and len(pos1_vids) >0 :
        videos.insert(0, {'videoId': int(pos1_vids[0]), 'rovScore': 100,
                          'pushFrom': config_.PUSH_FROM['position_insert'], 'abCode': config_.AB_CODE['position_insert']})
    if pos2_vids is not None and len(pos2_vids) >0 :
        videos.insert(1, {'videoId': int(pos2_vids[0]), 'rovScore': 100,
                          'pushFrom': config_.PUSH_FROM['position_insert'], 'abCode': config_.AB_CODE['position_insert']})

    return videos[:10]


def positon_duplicate(pos1_vids, pos2_vids, videos):
    s = set()
    if pos1_vids is not None and len(pos1_vids) >0:
        s.add(int(pos1_vids[0]))
    if pos2_vids is not None and len(pos2_vids) >0:
        s.add(int(pos2_vids[0]))

    l = []
    for i in range(len(videos)):
        if videos[i]['videoId'] in s:
            l.append(i)
    for i in l:
        videos.pop(i)
    return videos

def video_recommend(mid, uid, size, app_type, algo_type, client_info):
    """
    首页线上推荐逻辑
    :param mid: mid type-string
    :param uid: uid type-string
    :param size: 请求视频数量 type-int
    :param app_type: 产品标识  type-int
    :param algo_type: 算法类型  type-string
    :param client_info: 用户位置信息 {"country": "国家",  "province": "省份",  "city": "城市"}
    :return:
    """
    ab_code = config_.AB_CODE['initial']
    # ####### 多进程召回
    start_recall = time.time()
    # log_.info('====== recall')
    '''     
    cores = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes=cores)
    pool_recall = PoolRecall(app_type=app_type, mid=mid, uid=uid, ab_code=ab_code)
    _, last_rov_recall_key, _ = pool_recall.get_video_last_idx()
    pool_list = [
        # rov召回池
        pool.apply_async(pool_recall.rov_pool_recall, (size,)),
        # 流量池
        pool.apply_async(pool_recall.flow_pool_recall, (size,))
    ]
    recall_result_list = [p.get() for p in pool_list]
    pool.close()
    pool.join()
    '''
    recall_result_list = []
    pool_recall = PoolRecall(app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info)
    _, last_rov_recall_key, _ = pool_recall.get_video_last_idx()
    t = [gevent.spawn(pool_recall.rov_pool_recall, size), gevent.spawn(pool_recall.flow_pool_recall, size)]
    gevent.joinall(t)
    recall_result_list = [i.get() for i in t]

    end_recall = time.time()
    log_.info('mid: {}, uid: {}, recall: {}, execute time = {}ms'.format(
        mid, uid, recall_result_list, (end_recall - start_recall) * 1000))

    # ####### 排序
    start_rank = time.time()
    # log_.info('====== rank')
    data = {
        'rov_pool_recall': recall_result_list[0],
        'flow_pool_recall': recall_result_list[1]
    }
    rank_result = video_rank(data=data, size=size)
    end_rank = time.time()
    log_.info('mid: {}, uid: {}, rank_result: {}, execute time = {}ms'.format(
        mid, uid, rank_result, (end_rank - start_rank) * 1000))

    if not rank_result:
        # 兜底策略
        # log_.info('====== bottom strategy')
        start_bottom = time.time()
        rank_result = bottom_strategy(size=size, app_type=app_type, ab_code=ab_code)
        end_bottom = time.time()
        log_.info('mid: {}, uid: {}, bottom strategy result: {}, execute time = {}ms'.format(
            mid, uid, rank_result, (end_bottom - start_bottom) * 1000))

    return rank_result, last_rov_recall_key


def ab_test_op(rank_result, ab_code_list, app_type, mid, uid, **kwargs):
    """
    对排序后的结果 按照AB实验进行对应的分组操作
    :param rank_result: 排序后的结果
    :param ab_code_list: 此次请求参与的 ab实验组
    :param app_type: 产品标识
    :param mid: mid
    :param uid: uid
    :param kwargs: 其他参数
    :return:
    """
    # ####### 视频宽高比AB实验
    # 对内容精选进行 视频宽高比分发实验
    if config_.AB_CODE['w_h_rate'] in ab_code_list and app_type in config_.AB_TEST.get('w_h_rate', []):
        rank_result = video_rank_by_w_h_rate(videos=rank_result)
        log_.info('app_type: {}, mid: {}, uid: {}, rank_by_w_h_rate_result: {}'.format(
            app_type, mid, uid, rank_result))

    # 按position位置排序
    if config_.AB_CODE['position_insert'] in ab_code_list and app_type in config_.AB_TEST.get('position_insert', []):
        rank_result = video_position_recommend(mid, uid, app_type, rank_result)
        print('===========================')
        print(rank_result)
        log_.info('app_type: {}, mid: {}, uid: {}, rank_by_position_insert_result: {}'.format(
            app_type, mid, uid, rank_result))

    return rank_result


def update_redis_data(result, app_type, mid, last_rov_recall_key):
    """
    根据最终的排序结果更新相关redis数据
    :param result: 排序结果
    :param app_type: 产品标识
    :param mid: mid
    :param last_rov_recall_key: 用户上一次在rov召回池对应的位置 redis key
    :return: None
    """
    # ####### redis数据刷新
    try:
        # log_.info('====== update redis')
        # 预曝光数据同步刷新到Redis, 过期时间为0.5h
        redis_helper = RedisHelper()
        preview_key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(app_type, mid)
        preview_video_ids = [int(item['videoId']) for item in result]
        if preview_video_ids:
            # log_.error('key_name = {} \n values = {}'.format(preview_key_name, tuple(preview_video_ids)))
            redis_helper.add_data_with_set(key_name=preview_key_name, values=tuple(preview_video_ids), expire_time=30 * 60)
            log_.info('preview redis update success!')

        # 将此次获取的ROV召回池config_.K末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
        rov_recall_video = [item['videoId'] for item in result[:config_.K]
                            if item['pushFrom'] == config_.PUSH_FROM['rov_recall']]
        if len(rov_recall_video) > 0:
            if not redis_helper.get_score_with_value(key_name=config_.UPDATE_ROV_KEY_NAME, value=rov_recall_video[-1]):
                redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[-1])
            log_.info('last video redis update success!')

        # 将此次分发的流量池视频，对 本地分发数-1 进行记录
        flow_recall_video = [item for item in result if item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
        if flow_recall_video:
            update_local_distribute_count(flow_recall_video)
            log_.info('update local distribute count success!')

    except Exception as e:
        log_.error("update redis data fail!")
        log_.error(traceback.format_exc())


def update_local_distribute_count(videos):
    """
    更新本地分发数
    :param videos: 视频列表 type-list [{'videoId':'', 'flowPool':'', 'distributeCount': '',
                                    'rovScore': '', 'pushFrom': 'flow_pool', 'abCode': self.ab_code}, ....]
    :return:
    """
    try:
        redis_helper = RedisHelper()
        for item in videos:
            key_name = '{}{}.{}'.format(config_.LOCAL_DISTRIBUTE_COUNT_PREFIX, item['videoId'], item['flowPool'])
            # 本地记录的分发数 - 1
            redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)

            # if redis_helper.key_exists(key_name=key_name):
            #     # 该视频本地有记录，本地记录的分发数 - 1
            #     redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)
            # else:
            #     # 该视频本地无记录，接口获取的分发数 - 1
            #     redis_helper.incr_key(key_name=key_name, amount=int(item['distributeCount']) - 1, expire_time=5 * 60)

    except Exception as e:
        log_.error('update_local_distribute_count error...')
        log_.error(traceback.format_exc())


def video_homepage_recommend(mid, uid, size, app_type, algo_type, client_info):
    """
    首页线上推荐逻辑
    :param mid: mid type-string
    :param uid: uid type-string
    :param size: 请求视频数量 type-int
    :param app_type: 产品标识  type-int
    :param algo_type: 算法类型  type-string
    :param client_info: 用户位置信息 {"country": "国家",  "province": "省份",  "city": "城市"}
    :return:
    """
    # 简单召回 - 排序 - 兜底
    rank_result, last_rov_recall_key = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
                                                       algo_type=algo_type, client_info=client_info)
    # ab-test
    result = ab_test_op(rank_result=rank_result,
                        ab_code_list=[config_.AB_CODE['w_h_rate'], config_.AB_CODE['position_insert']],
                        app_type=app_type, mid=mid, uid=uid)
    # redis数据刷新
    update_redis_data(result=result, app_type=app_type, mid=mid, last_rov_recall_key=last_rov_recall_key)

    return result


def video_relevant_recommend(video_id, mid, uid, size, app_type):
    """
    相关推荐逻辑
    :param video_id: 相关推荐的头部视频id
    :param mid: mid type-string
    :param uid: uid type-string
    :param size: 请求视频数量 type-int
    :param app_type: 产品标识  type-int
    :return: videos type-list
    """
    # videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type, algo_type='', client_info=None)
    # 简单召回 - 排序 - 兜底
    rank_result, last_rov_recall_key = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
                                                       algo_type='', client_info=None)
    # ab-test
    result = ab_test_op(rank_result=rank_result,
                        ab_code_list=[config_.AB_CODE['w_h_rate'], config_.AB_CODE['position_insert']],
                        app_type=app_type, mid=mid, uid=uid, video_id=video_id)
    # redis数据刷新
    update_redis_data(result=result, app_type=app_type, mid=mid, last_rov_recall_key=last_rov_recall_key)

    return result


if __name__ == '__main__':
    videos = [{'videoId': '12345', 'flowPool': '133#442#2', 'distributeCount': 10}]
    update_local_distribute_count(videos)