import time import traceback from datetime import date, timedelta, datetime from log import Log from db_helper import RedisHelper from config import set_config from utils import FilterVideos, get_videos_remain_view_count, get_videos_local_distribute_count, send_msg_to_feishu import gevent log_ = Log() config_ = set_config() class PoolRecall(object): """召回""" def __init__(self, app_type, client_info=None, mid='', uid='', ab_code=''): """ 初始化 :param app_type: 产品标识 type-int :param client_info: 用户位置信息 {"cityCode": "100000"} :param mid: mid type-string :param uid: uid type-string :param ab_code: ab_code type-int """ self.app_type = app_type self.mid = mid self.uid = uid self.ab_code = ab_code self.client_info = client_info self.redis_helper = RedisHelper() def rov_pool_recall(self, size=10): """从ROV召回池中获取视频""" # log_.info('====== rov pool recall') # 获取生效中的置顶视频 if self.client_info is None: # 无用户位置信息时，不获取置顶视频 top_video_ids, top_video_result = [], [] else: top_video_ids, top_video_result = self.get_top_videos() # log_.info('===top video result = {}'.format(top_video_ids)) # 获取修改过rov的视频 update_rov_video_ids, update_rov_result = self.get_update_rov_videos() # log_.info('update rov result = {}'.format(update_rov_video_ids)) # 与置顶视频去重 update_rov_video_ids_dup, update_rov_dup_result = [], [] for item in update_rov_result: if item['videoId'] not in top_video_ids: update_rov_video_ids_dup.append(item['videoId']) update_rov_dup_result.append(item) # 获取相关redis key, 用户上一次在rov召回池对应的位置 rov_pool_key, last_rov_recall_key, idx = self.get_video_last_idx() if not rov_pool_key: log_.info('ROV召回池中无视频') if not update_rov_result: return [] return update_rov_result rov_pool_recall_result = [] # 每次获取的视频数 get_size = size * 2 # 记录获取频次 freq = 0 while len(rov_pool_recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 st_get = time.time() data = self.redis_helper.get_data_zset_with_index(key_name=rov_pool_key, start=idx, end=idx + get_size - 1, with_scores=True) et_get = time.time() # log_.info('get data from rov pool redis: freq = {}, data = {}, execute time = {}ms'.format( # freq, data, (et_get - st_get) * 1000)) if not data: log_.info('ROV召回池中的视频已取完') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} video_ids = [] video_score = {} for value in data: video_id = int(value[0]) # 视频id在 生效中的置顶视频 或 修改过rov的视频 中，跳过 if video_id in update_rov_video_ids_dup or video_id in top_video_ids: continue video_ids.append(video_id) video_score[video_id] = value[1] # 过滤 filter_ = FilterVideos(app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos) ge.join() filtered_result = ge.get() # filtered_result = filter_.filter_videos() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': config_.PUSH_FROM['rov_recall'], 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] rov_pool_recall_result.extend(temp_result) else: # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天 self.redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=data[-1][0]) idx += get_size # 生效中的置顶视频、被修改rov视频、rov召回池视频 归并排序 if top_video_result: rov_pool_recall_result.extend(top_video_result) if update_rov_dup_result: rov_pool_recall_result.extend(update_rov_dup_result) rov_pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True) return rov_pool_recall_result[:size] def flow_pool_recall(self, size=10): """从流量池中获取视频""" # log_.info('====== flow pool recall') flow_pool_key = self.get_pool_redis_key('flow') flow_pool_recall_result = [] flow_pool_recall_videos = [] # 每次获取的视频数 get_size = size * 3 # 记录获取频次 freq = 0 idx = 0 while len(flow_pool_recall_result) < size: freq += 1 # 获取数据 st_get = time.time() data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key, start=idx, end=idx + get_size - 1, with_scores=True) et_get = time.time() # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format( # freq, data, (et_get - st_get) * 1000)) if not data: log_.info('流量池中的视频已取完') break # 将video_id 与 flow_pool, score做mapping整理 video_ids = [] video_mapping = {} video_score = {} for value in data: video_id, flow_pool = value[0].split('-') video_id = int(video_id) if video_id not in video_ids: video_ids.append(video_id) video_score[video_id] = value[1] if video_id not in video_mapping: video_mapping[video_id] = [flow_pool] else: video_mapping[video_id].append(flow_pool) # 过滤 filter_ = FilterVideos(app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos) ge.join() filtered_result = ge.get() #filtered_result = filter_.filter_videos() # 检查可分发数 if filtered_result: st_check = time.time() ge = gevent.spawn(self.check_video_counts, video_ids=filtered_result, flow_pool_mapping=video_mapping) ge.join() check_result, error_flag = ge.get() #check_result, error_flag = self.check_video_counts(video_ids=filtered_result, flow_pool_mapping=video_mapping) for item in check_result: video_id = int(item[0]) flow_pool = item[1] if video_id not in flow_pool_recall_videos: # 取其中一个 flow_pool 作为召回结果 # 添加视频源参数 pushFrom, abCode flow_pool_recall_result.append( {'videoId': video_id, 'flowPool': flow_pool, 'rovScore': video_score[video_id], 'pushFrom': config_.PUSH_FROM['flow_recall'], 'abCode': self.ab_code} ) # flow_pool_recall_result.append( # {'videoId': item[0], 'flowPool': item[1], 'distributeCount': item[2], # 'rovScore': video_score[item[0]], 'pushFrom': 'flow_pool', 'abCode': self.ab_code} # ) flow_pool_recall_videos.append(video_id) et_check = time.time() # log_.info('check result: result = {}, execute time = {}ms'.format( # check_result, (et_check - st_check) * 1000)) # 判断错误标记, True为错误 if error_flag: # 结束流量池召回 break idx += get_size return flow_pool_recall_result[:size] def check_video_counts(self, video_ids, flow_pool_mapping): """ 检查视频剩余可分发数 :param video_ids: 视频id type-list :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict :return: check_result, error_flag """ # flow_pool_key = self.get_pool_redis_key('flow') videos = [] check_result = [] for video_id in video_ids: video_id = int(video_id) for flow_pool in flow_pool_mapping.get(video_id, []): # 判断是否有本地分发记录 cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool) # 无记录 if cur_count is None: videos.append({'videoId': video_id, 'flowPool': flow_pool}) # 本地分发数 cur_count > 0 elif cur_count > 0: check_result.append((video_id, flow_pool)) # 本地分发数 cur_count <= 0，从所有的流量召回池移除 else: value = '{}-{}'.format(video_id, flow_pool) for item in config_.APP_TYPE: flow_pool_key = config_.FLOW_POOL_KEY_NAME_PREFIX + str(config_.APP_TYPE.get(item)) self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value) # 本次视频都有本地记录 if len(videos) == 0: error_flag = False return check_result, error_flag # 本地无记录视频，检查实时分发数 view_count_result, error_flag = get_videos_remain_view_count(app_type=self.app_type, videos=videos) # log_.info('error_flag = {}, view_count_result = {}'.format(error_flag, view_count_result)) # 判断返回的错误标记，True为错误 if error_flag: return check_result, error_flag # 从流量召回池移除视频videos # for item in videos: # value = '{}-{}'.format(item['videoId'], item['flowPool']) # self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value) redis_helper = RedisHelper() for item in view_count_result: try: # 接口超时，item[2]可能为None remain_count = int(item[2]) except Exception as e: log_.error('remain_count type error...') log_.error(traceback.format_exc()) continue if remain_count > 0: # viewCount > 0 check_result.append(item) # 将分发数更新到本地记录 key_name = '{}{}.{}'.format(config_.LOCAL_DISTRIBUTE_COUNT_PREFIX, item[0], item[1]) redis_helper.setnx_key(key_name=key_name, value=remain_count, expire_time=5 * 60) else: # viewCount <= 0 # 从流量召回池移除 value = '{}-{}'.format(item[0], item[1]) for item in config_.APP_TYPE: flow_pool_key = config_.FLOW_POOL_KEY_NAME_PREFIX + str(config_.APP_TYPE.get(item)) self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value) return check_result, error_flag def get_pool_redis_key(self, pool_type): """ 拼接key :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池} :return: key_name """ if pool_type == 'rov': # appType = 6 if self.app_type == config_.APP_TYPE['SHORT_VIDEO']: # 获取当前所在小时 redis_date = datetime.now().hour # 判断热度列表是否更新，未更新则使用前一小时的热度列表 key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, redis_date) if self.redis_helper.key_exists(key_name): return key_name, redis_date else: key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, redis_date - 1) return key_name, redis_date - 1 # 其他 else: # 判断热度列表是否更新，未更新则使用前一天的热度列表 key_name = config_.RECALL_KEY_NAME_PREFIX + time.strftime('%Y%m%d') if self.redis_helper.key_exists(key_name): redis_date = date.today().strftime('%Y%m%d') else: redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') key_name = config_.RECALL_KEY_NAME_PREFIX + redis_date # if not self.redis_helper.key_exists(key_name): # return None, None # 判断当前时间是否晚于rov召回池更新时间 + 1h，发送消息到飞书 now_h = datetime.now().hour now_m = datetime.now().minute feishu_text = '预发布环境 —— 今日ROV召回池数据未按时更新，请及时查看解决。' if now_h == config_.ROV_UPDATE_H + 1 and now_m > config_.ROV_UPDATE_MINUTE: send_msg_to_feishu(feishu_text) elif now_h > config_.ROV_UPDATE_H + 2: send_msg_to_feishu(feishu_text) return key_name, redis_date elif pool_type == 'flow': return config_.FLOW_POOL_KEY_NAME_PREFIX + str(self.app_type) else: log_.error('pool type error') return None, None def get_video_last_idx(self): """获取用户上一次在rov召回池对应的位置""" rov_pool_key, redis_date = self.get_pool_redis_key('rov') if not rov_pool_key: return None, None, None last_rov_recall_key = config_.LAST_VIDEO_FROM_ROV_POOL_PREFIX + '{}.{}.{}'.format( self.app_type, self.mid, redis_date) value = self.redis_helper.get_data_from_redis(last_rov_recall_key) if value: idx = self.redis_helper.get_index_with_data(rov_pool_key, value) if not idx: idx = 0 else: idx += 1 else: idx = 0 return rov_pool_key, last_rov_recall_key, idx def get_update_rov_videos(self): """ 获取修改ROV的视频 :return: update_rov_video_ids, update_rov_result """ try: # 获取修改过ROV的视频 redis_helper = RedisHelper() data = redis_helper.get_data_zset_with_index(key_name=config_.UPDATE_ROV_KEY_NAME, start=0, end=-1, with_scores=True) # 获取视频id,并转换类型为int,将videoId和score做mapping，并存储为key-value{videoId: score} if data is None: return [], [] video_ids = [] video_score = {} for value in data: video_id = int(value[0]) video_ids.append(video_id) video_score[video_id] = value[1] # 过滤 filter_ = FilterVideos(app_type=self.app_type, video_ids=video_ids, mid=self.mid, uid=self.uid) ge = gevent.spawn(filter_.filter_videos) ge.join() filtered_result = ge.get() # 添加视频源参数 pushFrom, abCode update_rov_video_ids, update_rov_result = [], [] if not filtered_result: return update_rov_video_ids, update_rov_result for item in filtered_result: video_id = int(item) rov_score = video_score.get(video_id) if rov_score is None: continue update_rov_video_ids.append(video_id) update_rov_result.append({'videoId': video_id, 'rovScore': rov_score, 'pushFrom': config_.PUSH_FROM['rov_recall'], 'abCode': self.ab_code}) return update_rov_video_ids, update_rov_result except Exception as e: log_.error(traceback.format_exc()) return [], [] def get_top_videos(self): """ 获取置顶视频 :return: top_video_ids, top_video_result """ try: # 获取生效中的置顶视频列表 redis_helper = RedisHelper() data = redis_helper.get_data_from_redis(key_name=config_.TOP_VIDEO_LIST_KEY_NAME) # log_.info('===1=== {}'.format(data)) if data is None: return [], [] # 获取视频id,并转换类型为int,将videoId和score做mapping，并存储为key-value{videoId: score} video_ids = [] video_info = {} for item in eval(data): # log_.info('=== uid: {}, item === {}'.format(self.uid, item)) video_id = int(item['videoId']) # 判断 视频推荐区域与用户地址信息 是否匹配，市级别 city_code_list = item['cityCode'].split(',') # log_.info('=== uid: {}, city_code_list: {},{}, cityCode: {},{} === '.format( # self.uid, city_code_list, type(city_code_list[0]), self.client_info.get('cityCode'), # type(self.client_info.get('cityCode')))) if self.client_info.get('cityCode') in city_code_list or config_.ALL_AREA_CODE in city_code_list: # log_.info('=== uid: {}, ===2=== {}, ===3=== video_id: {}, cityCode: {}'.format( # self.uid, self.client_info.get('cityCode'), video_id, city_code_list)) if video_id in video_ids: # 如果video_id有重复，score保留大分值 if item['score'] > video_info[video_id].get('score'): video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list} else: continue else: video_ids.append(video_id) video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list} else: continue # 过滤 filter_ = FilterVideos(app_type=self.app_type, video_ids=video_ids, mid=self.mid, uid=self.uid) ge = gevent.spawn(filter_.filter_videos) ge.join() filtered_result = ge.get() # 添加视频源参数 pushFrom = 'op_manual', abCode top_video_ids, top_video_result = [], [] if not filtered_result: return top_video_ids, top_video_result for item in filtered_result: video_id = int(item) item_info = video_info.get(video_id) if item_info is None: continue top_video_ids.append(video_id) top_video_result.append({'videoId': video_id, 'rovScore': item_info.get('score'), 'pushFrom': config_.PUSH_FROM['top'], 'abCode': self.ab_code}) return top_video_ids, top_video_result except Exception as e: log_.error(traceback.format_exc()) return [], []