rov-offline/flowpool_data_update.py

import datetime
import random
import time
import os
import traceback
import random

from config import set_config
from utils import request_post, filter_video_status, send_msg_to_feishu, filter_video_status_app, \
    filter_political_videos
from log import Log
from db_helper import RedisHelper
from odps import ODPS

config_, _ = set_config()
log_ = Log()


def get_videos_from_flow_pool(app_type, size=1000):
    """
    从流量池获取视频，循环获取，直到返回数据为None结束
    :param app_type: 产品标识 type-int
    :param size: 每次获取视频数量，type-int，默认1000
    :return: videos  [{'videoId': 1111, 'flowPool': ''}, ...]
    """
    # 获取批次标识，利用首次获取数据时间戳为标记
    batch_flag = int(time.time())
    request_data = {'appType': app_type, 'batchFlag': batch_flag, 'size': size}
    videos = []
    while True:
        result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data)
        if result is None:
            break
        if result['code'] != 0:
            log_.info('batch_flag: {}, 获取流量池视频失败'.format(batch_flag))
            break
        if not result['data']:
            break
        videos.extend(result['data'])
    return videos


def get_videos_remain_view_count(video_info_list):
    """
    获取视频在流量池中的剩余可分发数，并存入对应的redis中
    :param app_type: 产品标识 type-int
    :param video_info_list: 视频信息 (视频id, 流量池标记) type-list，[(video_id, flow_pool), ...]
    :return: data type-list，[(video_id, flow_pool, view_count), ...]
    """
    redis_helper = RedisHelper()
    if not video_info_list:
        return

    # 每次请求10个
    for i in range(len(video_info_list)//10 + 1):
        remain_st_time = time.time()
        videos = [{'videoId': info[0], 'flowPool': info[1]} for info in video_info_list[i*10:(i+1)*10]]
        request_data = {'videos': videos}
        result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL,
                              request_data=request_data, timeout=(0.5, 3))
        log_.info(f"i = {i}, expend time = {(time.time()-remain_st_time)*1000}")
        if result is None:
            continue
        if result['code'] != 0:
            log_.error('获取视频在流量池中的剩余可分发数失败')
            continue
        for item in result['data']:
            if item['distributeCount'] is None:
                continue
            distribute_count = int(item['distributeCount'])
            if distribute_count > 0:
                # 将分发数更新到本地记录
                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
                redis_helper.set_data_to_redis(key_name=key_name, value=distribute_count, expire_time=15 * 60)
            else:
                # 将本地记录删除
                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
                redis_helper.del_keys(key_name=key_name)
                # 从流量召回池移除
                value = '{}-{}'.format(item['videoId'], item['flowPool'])
                for type_name in config_.APP_TYPE:
                    flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}"
                    redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
                    quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}" \
                                          f":{config_.QUICK_FLOW_POOL_ID}"
                    redis_helper.remove_value_from_set(key_name=quick_flow_pool_key, values=(value, ))


def get_flow_pool_recommend_config(flow_pool_id):
    """获取流量池推荐分发配置"""
    result = request_post(request_url=config_.GET_FLOW_POOL_RECOMMEND_CONFIG_URL)
    if result is None:
        return None
    if result['code'] != 0:
        return None
    flow_pool_distribute_config = result['data'].get('flowPoolDistributeConfig')
    if flow_pool_distribute_config:
        if int(eval(flow_pool_distribute_config).get('flowPoolId')) == flow_pool_id:
            return eval(eval(flow_pool_distribute_config).get('distributeRate'))
        else:
            return None
    else:
        return None


def online_flow_pool_data_to_redis(app_type, video_ids_set, video_info_data):
    redis_helper = RedisHelper()
    for tag, values in video_ids_set.items():
        if tag == 'quick_flow_pool':
            key_name_video_ids = \
                f"{config_.QUICK_FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
            key_prefix_video_info = \
                f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
        else:
            key_name_video_ids = f"{config_.FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}"
            key_prefix_video_info = f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}"
        # 如果key已存在，删除key
        if redis_helper.key_exists(key_name=key_name_video_ids):
            redis_helper.del_keys(key_name=key_name_video_ids)
        # 写入redis
        if len(values) > 0:
            redis_helper.add_data_with_set(key_name=key_name_video_ids, values=values, expire_time=3600)
            info_values = video_info_data[tag]
            if len(info_values) > 0:
                for video_id, info_value in info_values.items():
                    key_name_video_info = f"{key_prefix_video_info}:{video_id}"
                    # 如果key已存在，删除key
                    if redis_helper.key_exists(key_name=key_name_video_info):
                        redis_helper.del_keys(key_name=key_name_video_info)
                    if len(info_value) > 0:
                        redis_helper.add_data_with_set(key_name=key_name_video_info, values=info_value,
                                                       expire_time=3600)


def get_flow_pool_data(app_type, video_info_list):
    """
    获取流量池可分发视频，并将结果上传Redis
    :param app_type: 产品标识 type-int
    :return: None
    """
    try:
        # 从流量池获取数据
        videos = get_videos_from_flow_pool(app_type=app_type)
        if len(videos) <= 0:
            log_.info('流量池中无需分发的视频')
            return video_info_list
        # video_id 与 flow_pool 进行mapping
        video_ids = set()
        log_.info('流量池中视频数：{}'.format(len(videos)))
        mapping = {}
        for video in videos:
            video_id = video['videoId']
            video_ids.add(video_id)
            if video_id in mapping:
                mapping[video_id].append(video['flowPool'])
            else:
                mapping[video_id] = [video['flowPool']]

        # 对视频状态进行过滤
        if app_type == config_.APP_TYPE['APP']:
            filtered_videos = filter_video_status_app(list(video_ids))
        else:
            filtered_videos = filter_video_status(list(video_ids))
        log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))

        # 涉政视频过滤
        if app_type not in config_.POLITICAL_RECOMMEND_APP_TYPE_LIST:
            filtered_videos = filter_political_videos(video_ids=filtered_videos)

        if not filtered_videos:
            log_.info('流量池中视频状态不符合分发')
            return video_info_list

        # 上传数据到redis
        redis_data = set()
        quick_flow_pool_redis_data = set()

        # 用于线上流量池标记补充
        video_ids_set = {'quick_flow_pool': set(), 'flow_pool': set()}
        video_info_data = {'quick_flow_pool': {}, 'flow_pool': {}}

        for video_id in filtered_videos:
            quick_flow_pool_list = set()
            flow_pool_list = set()
            for flow_pool in mapping.get(video_id):
                # 判断是否为快速曝光流量池视频
                value = '{}-{}'.format(video_id, flow_pool)
                flow_pool_id = int(flow_pool.split('#')[0])  # flowPool: 流量池ID#分级ID#级别Level#生命周期ID
                if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
                    quick_flow_pool_redis_data.add(value)
                    video_ids_set['quick_flow_pool'].add(video_id)
                    quick_flow_pool_list.add(flow_pool)
                else:
                    redis_data.add(value)
                    video_ids_set['flow_pool'].add(video_id)
                    flow_pool_list.add(flow_pool)
                video_info = (video_id, flow_pool)
                if video_info not in video_info_list:
                    video_info_list.append(video_info)
            if len(quick_flow_pool_list) > 0:
                video_info_data['quick_flow_pool'][video_id] = quick_flow_pool_list
            if len(flow_pool_list) > 0:
                video_info_data['flow_pool'][video_id] = flow_pool_list

        # 快速曝光流量池视频写入redis
        redis_helper = RedisHelper()
        quick_flow_pool_key_name = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
        # 如果key已存在，删除key
        if redis_helper.key_exists(quick_flow_pool_key_name):
            redis_helper.del_keys(quick_flow_pool_key_name)
        # 写入redis
        if quick_flow_pool_redis_data:
            log_.info(f"quick_flow_pool_redis_data = {quick_flow_pool_redis_data}")
            redis_helper.add_data_with_set(key_name=quick_flow_pool_key_name, values=quick_flow_pool_redis_data,
                                           expire_time=24 * 3600)
            # 快速流量池分发概率存入redis
            distribute_rate_key_name = f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
            distribute_rate = get_flow_pool_recommend_config(flow_pool_id=config_.QUICK_FLOW_POOL_ID)
            if distribute_rate is not None:
                redis_helper.set_data_to_redis(key_name=distribute_rate_key_name, value=distribute_rate,
                                               expire_time=15 * 60)

        # 普通流量池视频写入redis
        flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}"
        # 如果key已存在，删除key
        if redis_helper.key_exists(flow_pool_key_name):
            redis_helper.del_keys(flow_pool_key_name)
        # 写入redis
        if redis_data:
            redis_helper.add_data_with_set(key_name=flow_pool_key_name, values=redis_data, expire_time=24 * 3600)

        # 用于线上视频补充流量池标记的数据写入redis
        online_flow_pool_data_to_redis(app_type=app_type, video_ids_set=video_ids_set, video_info_data=video_info_data)

        log_.info('data to redis finished!')

        return video_info_list

    except Exception as e:
        log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
            app_type, e, 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='rov-offline{} - 流量池更新失败, appType: {}, exception: {}'.format(config_.ENV_TEXT, app_type, e)
        )
        return video_info_list


def get_data_from_odps(project, sql):
    """检查数据是否准备好"""
    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:
        with odps.execute_sql(sql=sql).open_reader() as reader:
            data_df = reader.to_pandas()
    except Exception as e:
        data_df = None
    return data_df


def predict_18_19(app_type):
    log_.info(f'app_type = {app_type}')
    now = datetime.datetime.now()
    log_.info(f"now = {datetime.datetime.strftime(now, '%Y-%m-%d %H:%M:%S')}")
    # create_time = datetime.datetime.strftime(now - datetime.timedelta(hours=24), '%Y-%m-%d %H:%M:%S')
    create_time = '2022-04-22 16:40:00'
    if app_type == config_.APP_TYPE['LAO_HAO_KAN_VIDEO']:
        sql = f"SELECT video_id FROM videoods.movie_store_video_allow_list " \
              f"WHERE allow_list_type=1 AND create_time>='{create_time}'"
    # elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
    #     sql = f"SELECT video_id FROM videoods.movie_store_video_allow_list " \
    #           f"WHERE allow_list_type=0 AND " \
    #           f"video_id NOT IN (" \
    #           f"SELECT video_id FROM videoods.movie_store_video_allow_list WHERE allow_list_type=1" \
    #           f") AND " \
    #           f"create_time>='{create_time}'"
    else:
        sql = ""
    data_df = get_data_from_odps(project='videoods', sql=sql)
    if data_df is not None:
        video_ids = [int(video_id) for video_id in data_df['video_id'].to_list()]
        log_.info(f'video_ids count = {len(video_ids)}')
        # 对视频状态进行过滤
        filtered_videos = filter_video_status(list(video_ids))
        log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))
        if not filtered_videos:
            log_.info('流量池中视频状态不符合分发')
            return None
        # 预测
        video_score = get_score(filtered_videos)
        log_.info('predict finished!')
        # 上传数据到redis
        redis_data = {}
        for i in range(len(video_score)):
            video_id = filtered_videos[i]
            score = video_score[i]
            redis_data[video_id] = score
        key_name = f'{config_.FLOWPOOL_KEY_NAME_PREFIX}{app_type}'
        redis_helper = RedisHelper()
        # 如果key已存在，删除key
        if redis_helper.key_exists(key_name):
            redis_helper.del_keys(key_name)
        # 写入redis
        redis_helper.add_data_with_zset(key_name=key_name, data=redis_data, expire_time=24 * 3600)
        log_.info('data to redis finished!')


def get_score_19(video_ids):
    data = {}
    step = round(100.0 / len(video_ids), 3)
    for i, video_id in enumerate(video_ids):
        score = 100 - i * step
        data[video_id] = score
    return data


def predict_19(app_type):
    log_.info(f'app_type = {app_type}')
    now = datetime.datetime.now()
    log_.info(f"now = {datetime.datetime.strftime(now, '%Y-%m-%d %H:%M:%S')}")
    sql_create_time = datetime.datetime.strftime(now - datetime.timedelta(days=30), '%Y-%m-%d %H:%M:%S')
    if sql_create_time < '2022-04-22 16:40:00':
        sql_create_time = '2022-04-22 16:40:00'
    sql = f"SELECT video_id, create_time FROM videoods.movie_store_video_allow_list_final " \
          f"WHERE create_time>='{sql_create_time}'" \
          f"ORDER BY create_time DESC;"
    data_df = get_data_from_odps(project='videoods', sql=sql)

    if data_df is not None:
        video_ids = [int(video_id) for video_id in data_df['video_id'].to_list()]
        log_.info(f'video_ids count = {len(video_ids)}')
        # 预测
        video_score = get_score_19(video_ids=video_ids)
        # 对视频状态进行过滤
        filtered_videos = filter_video_status(list(video_ids))
        log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))
        if not filtered_videos:
            log_.info('流量池中视频状态不符合分发')
            return None
        # 上传数据到redis
        data = {}
        for video_id in filtered_videos:
            score = video_score[video_id]
            data[video_id] = score
        log_.info('predict finished!')
        key_name = f'{config_.FLOWPOOL_KEY_NAME_PREFIX}{app_type}'
        redis_helper = RedisHelper()
        # 如果key已存在，删除key
        if redis_helper.key_exists(key_name):
            redis_helper.del_keys(key_name)
        # 写入redis
        redis_helper.add_data_with_zset(key_name=key_name, data=data, expire_time=24 * 3600)
        log_.info('data to redis finished!')


if __name__ == '__main__':
    st_time = time.time()
    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
    log_.info('flow pool predict start...')
    video_info_list = []
    for app_name, app_type in config_.APP_TYPE.items():
        log_.info('{} predict start...'.format(app_name))
        if app_type == config_.APP_TYPE['LAO_HAO_KAN_VIDEO']:
            # predict_18_19(app_type=app_type)
            continue
        elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
            # predict_19(app_type=app_type)
            continue
        else:
            video_info_list = get_flow_pool_data(app_type=app_type, video_info_list=video_info_list)
        log_.info('{} predict end...'.format(app_name))

    # 更新剩余分发数
    log_.info(f"video_info_list count = {len(video_info_list)}")
    get_videos_remain_view_count(video_info_list)
    log_.info('flow pool predict end...')
    log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")