import datetime import random import time import os import traceback import random import json from my_config import set_config from my_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 = [] retry = 0 while True: print(config_.GET_VIDEOS_FROM_POOL_URL) result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data) if result is None: if retry > 2: break retry += 1 continue if result['code'] != 0: log_.info('batch_flag: {}, 获取流量池视频失败'.format(batch_flag)) if retry > 2: break retry += 1 continue if not result['data']: if retry > 2: break retry += 1 continue 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个 bu_fen_fa_cnt = 0 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=25 * 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: for level in range(1, 7): flow_pool_key = \ f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{config_.APP_TYPE.get(type_name)}:{level}" 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, )) bu_fen_fa_cnt = bu_fen_fa_cnt + 1 log_.info(f"新增加不分发过滤前后整体数量: {len(video_info_list)}:{str(bu_fen_fa_cnt)}") 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, flow_pool_id_list): """ 获取流量池可分发视频，并将结果上传Redis :param app_type: 产品标识 type-int :return: None """ try: # 从流量池获取数据 videos = get_videos_from_flow_pool(app_type=app_type) log_.info(f"app_type: {app_type}") # log_.info(f"videos: {videos}") if len(videos) <= 0: log_.info('流量池中无需分发的视频') return video_info_list # video_id 与 flow_pool, level 进行mapping video_ids = set() log_.info('流量池中视频数：{}'.format(len(videos))) mapping = {} for video in videos: flow_pool_id = video['flowPoolId'] # 召回使用的切分ID是在这里做的。流量池中的视频是不区分ID的，也不区分层。 if int(flow_pool_id) not in flow_pool_id_list: continue # print(f"flow_pool_id: {flow_pool_id}") video_id = video['videoId'] video_ids.add(video_id) item_info = {'flowPool': video['flowPool'], 'level': video['level']} if video_id in mapping: mapping[video_id].append(item_info) else: mapping[video_id] = [item_info] log_.info(f"需更新流量池视频数: {len(video_ids)}") # 对视频状态进行过滤 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 quick_flow_pool_redis_data = set() # 普通流量池视频按照层级存储 redis_data = dict() for video_id in filtered_videos: for item in mapping.get(video_id): flow_pool = item['flowPool'] level = item['level'] # 判断是否为快速曝光流量池视频 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) else: if level not in redis_data: redis_data[level] = set() redis_data[level].add(value) video_info = (video_id, flow_pool) if video_info not in video_info_list: video_info_list.append(video_info) # 快速曝光流量池视频写入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 - 分层存储 level_list = [] for level, videos in redis_data.items(): log_.info(f"level: {level}, videos_count: {len(videos)}") level_list.append(level) flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}" # 如果key已存在，删除key if redis_helper.key_exists(flow_pool_key_name): redis_helper.del_keys(flow_pool_key_name) # 写入redis if videos: redis_helper.add_data_with_set(key_name=flow_pool_key_name, values=videos, expire_time=24 * 3600) result = redis_helper.get_data_from_set(flow_pool_key_name) if not result: result = [] size = len(result) log_.info(f'写入成功key={flow_pool_key_name}:{size}') # 删除此时不存在的level key for i in range(1, 7): if i not in level_list: flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{i}" # 如果key已存在，删除key if redis_helper.key_exists(flow_pool_key_name): redis_helper.del_keys(flow_pool_key_name) log_.info(f'删除不存在key={flow_pool_key_name}') 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 print_check_log(s): log_.info(s) apptype_list = ["0", "4", "5", "21", "3", "6"] level_list = ["1", "2", "3", "4"] date_cur = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") for app in apptype_list: for level in level_list: key = "flow:pool:level:item:" + app + ":" + level result = redis_helper.get_data_from_set(key) if not result: result = [] size = len(result) log_str = "\t".join([date_cur, app, level, str(size)]) log_.info(log_str) if __name__ == '__main__': st_time = time.time() # 为避免第一个app_type获取数据不全，等待1min time.sleep(60) app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']] log_.info('flow pool predict start...') # 获取对应流量池id列表 redis_helper = RedisHelper() flow_pool_abtest_config = redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_ABTEST_KEY_NAME) if flow_pool_abtest_config is not None: flow_pool_abtest_config = json.loads(flow_pool_abtest_config) else: flow_pool_abtest_config = {} flow_pool_id_list = flow_pool_abtest_config.get('experimental_flow_set_level', []) 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']: # continue # elif app_type == config_.APP_TYPE['ZUI_JING_QI']: # continue # else: video_info_list = get_flow_pool_data(app_type=app_type, video_info_list=video_info_list, flow_pool_id_list=flow_pool_id_list) log_.info('{} predict end...'.format(app_name)) # 更新剩余分发数 log_.info(f"video_info_list count = {len(video_info_list)}") print_check_log("--------------------前序日志--------------------") 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") print_check_log("--------------------后续日志--------------------") # python flowpool_data_update_with_level.py 测试环境必须手动执行python 才能有数据