import time import traceback import random 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, request_id, app_type, client_info=None, mid='', uid='', ab_code='', rule_key='', data_key='', no_op_flag=False, params=None): """ 初始化 :param request_id: request_id :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 :param params: """ self.request_id = request_id self.app_type = app_type self.mid = mid self.uid = uid self.ab_code = ab_code self.client_info = client_info self.rule_key = rule_key self.data_key = data_key self.no_op_flag = no_op_flag self.redis_helper = RedisHelper(params=params) def copy_redis_zset_data(self, from_key_name, to_key_name): # 获取from_key_name中的数据 records = self.redis_helper.get_data_zset_with_index(key_name=from_key_name, start=0, end=-1, with_scores=True) if records is not None: data = {} for video_id, score in records: data[int(video_id)] = score # 重新写入 if self.redis_helper.key_exists(to_key_name): self.redis_helper.del_keys(key_name=to_key_name) self.redis_helper.add_data_with_zset(key_name=to_key_name, data=data, expire_time=1*3600) return True else: return False def update_mid_data(self, h_recall_mid_key, h_record_key, key_prefix): # 判断当前小时的小时级列表是否更新 now_date = datetime.today() h = datetime.now().hour now_dt = datetime.strftime(now_date, '%Y%m%d') now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}" if self.redis_helper.key_exists(key_name=now_h_recall_key): flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key) if flag: value = {'date': now_dt, 'h': h} self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=1*3600) else: if h == 0: redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d') redis_h = 23 else: redis_dt = now_dt redis_h = h - 1 now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{redis_dt}.{redis_h}" flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key) if flag: value = {'date': redis_dt, 'h': redis_h} self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=1*3600) def get_mid_h_key(self, province_code, key_flag=''): if key_flag == 'region_24h': # mid对应小时级视频列表 redis-key 地域分组相对24h h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}" # 判断mid对应小时级视频列表 时间记录 h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}" elif key_flag in ['24h', 'day_24h']: # mid对应小时级视频列表 redis-key 相对24h h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}" # 判断mid对应小时级视频列表 时间记录 h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}" else: # mid对应小时级视频列表 redis-key h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX}{self.app_type}.{self.mid}" # 判断mid对应小时级视频列表 时间记录 h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX}{self.app_type}.{self.mid}" # 列表存储 redis-key prefix if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]: if key_flag == 'region_24h': if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'): key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{province_code}." else: key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{province_code}." elif key_flag == 'day_24h': key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{province_code}." else: key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{province_code}." elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]: key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_24H elif key_flag == '24h': key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_24H_H else: key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H if not self.redis_helper.key_exists(key_name=h_record_key): # ###### 记录key不存在，copy列表，更新记录 self.update_mid_data(h_recall_mid_key=h_recall_mid_key, h_record_key=h_record_key, key_prefix=key_prefix) # return h_recall_mid_key else: # ###### 记录key存在，判断date, h now_date = datetime.today() h = datetime.now().hour # 获取记录的date, h record = self.redis_helper.get_data_from_redis(key_name=h_record_key) record_dt = eval(record).get('date') record_h = eval(record).get('h') now_dt = datetime.strftime(now_date, '%Y%m%d') if record_dt == now_dt and int(record_h) == h: # 已获取当前小时数据 pass # return h_recall_mid_key elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23): # 记录的h - 当前h = 1，判断当前h数据是否已更新 now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}" # if not self.redis_helper.key_exists(key_name=now_h_recall_key): # 未更新 # return h_recall_mid_key if self.redis_helper.key_exists(key_name=now_h_recall_key): # 已更新，重新获取更新mid对应列表及记录 # self.redis_helper.del_keys(key_name=h_recall_mid_key) # self.redis_helper.del_keys(key_name=h_record_key) flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key) if flag: new_record = {'date': now_dt, 'h': h} self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(new_record), expire_time=2*3600) # return h_recall_mid_key else: self.update_mid_data(h_recall_mid_key=h_recall_mid_key, h_record_key=h_record_key, key_prefix=key_prefix) # return h_recall_mid_key return h_recall_mid_key def rov_pool_recall_by_h(self, size=10, expire_time=24*3600): """ 从小时级更新ROV召回池中获取视频 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :return: """ start_time = time.time() # 获取provinceCode province_code = self.client_info.get('provinceCode', '-1') if province_code == '': province_code = '-1' if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]: push_from = config_.PUSH_FROM['rov_recall_24h'] else: push_from = config_.PUSH_FROM['rov_recall_h'] # 获取mid对应的小时级列表redis-key h_recall_mid_key = self.get_mid_h_key(province_code=province_code) if not self.redis_helper.key_exists(h_recall_mid_key): recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) else: # 过滤的视频 fil_video_ids = [] recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key, start=(freq - 1) * get_size, end=freq * get_size - 1, with_scores=True) if not data: # log_.info('小时级更新视频已取完') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.ab_code, province_code) ge.join() filtered_result = ge.get() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] recall_result.extend(temp_result) fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result]))) else: fil_video_ids.extend(video_ids) # 将被过滤的视频进行移除 for value in fil_video_ids: self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value) # 判断获取到的小时级数据数量 if len(recall_result) < size: # 补充数据 rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) # 去重合并 now_video_ids = [item.get('videoId') for item in recall_result] for video in rov_recall_result: vid = video.get('videoId') if vid not in now_video_ids: recall_result.append(video) now_video_ids.append(vid) if len(recall_result) >= size: break else: continue log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'rov_pool_recall_by_h', 'executeTime': (time.time() - start_time) * 1000 }) return recall_result[:size] def rov_pool_recall_by_day(self, size=4, expire_time=24*3600): """ 从天级规则更新列表中获取视频 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :return: """ start_time = time.time() # 获取天级规则更新列表相关redis key, 用户上一次在天级规则更新列表对应的位置 rule_key_name, last_rule_day_recall_key, idx = self.get_video_last_idx_day() # 获取天级规则更新列表数据 if not rule_key_name: # log_.info('天级规则更新列表中无视频') recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) else: recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=rule_key_name, start=idx, end=idx + get_size - 1, with_scores=True) if not data: # log_.info('天级规则更新视频已取完') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, 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() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': config_.PUSH_FROM['rov_recall_day'], 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] recall_result.extend(temp_result) else: # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天 self.redis_helper.set_data_to_redis(key_name=last_rule_day_recall_key, value=data[-1][0], expire_time=expire_time) idx += get_size # 判断获取到的天级规则数据数量 if len(recall_result) < size: # 补充数据 rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) # 去重合并 now_video_ids = [item.get('videoId') for item in recall_result] for video in rov_recall_result: vid = video.get('videoId') if vid not in now_video_ids: recall_result.append(video) now_video_ids.append(vid) if len(recall_result) >= size: break else: continue log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'rov_pool_recall_by_day', 'executeTime': (time.time() - start_time) * 1000 }) return recall_result[:size] def rov_pool_recall(self, size=10, expire_time=24*3600, video_type='', push_from=config_.PUSH_FROM['rov_recall']): """ 从ROV召回池中获取视频 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :param video_type: 视频列表类别 :param push_from: 视频来源标记 :return: """ start_time = time.time() # 获取生效中的置顶视频 if self.no_op_flag: top_video_ids, top_video_result = [], [] elif 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的视频 if self.no_op_flag: update_rov_video_ids, update_rov_result = [], [] else: 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(video_type=video_type) if not rov_pool_key: # log_.info('ROV召回池中无视频') if (not update_rov_dup_result) and (not top_video_result): return [] rov_pool_recall_result = top_video_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] rov_pool_recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 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(request_id=self.request_id, 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': push_from, '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天 if self.mid: # mid为空时，不做记录 self.redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=data[-1][0], expire_time=expire_time) 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) log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'rov_pool_recall', 'executeTime': (time.time() - start_time) * 1000 }) return rov_pool_recall_result[:size] def flow_pool_recall(self, size=10, flow_pool_id=None): """从流量池中获取视频""" start_time = time.time() # 获取provinceCode province_code = self.client_info.get('provinceCode', '-1') if province_code == '': province_code = '-1' flow_pool_key = self.get_pool_redis_key('flow', flow_pool_id=flow_pool_id) print(flow_pool_key) flow_pool_recall_result = [] flow_pool_recall_videos = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 idx = 0 while len(flow_pool_recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_FLOW_POOL: break # 获取数据 # 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: try: video_id, flow_pool = value[0].split('-') except Exception as e: log_.error({ 'request_id': self.request_id, 'app_type': self.app_type, 'flow_pool_value': value }) continue 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(request_id=self.request_id, app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos, pool_type='flow', province_code=province_code) ge.join() filtered_result = ge.get() # 检查可分发数 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 = ge.get() log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'app_type': self.app_type, 'mid': self.mid, 'uid': self.uid, 'operation': 'check_video_counts', 'executeTime': (time.time() - st_check) * 1000 }) 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_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 log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'flow_pool_recall', 'executeTime': (time.time() - start_time) * 1000 }) 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}) continue # 本地分发数 cur_count > 0 elif cur_count > 0: check_result.append((video_id, flow_pool)) # 本地分发数 cur_count <= 0，从所有的流量召回池移除，删除本地分发记录key else: remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}" self.redis_helper.del_keys(remain_count_key) value = '{}-{}'.format(video_id, flow_pool) for item in config_.APP_TYPE: flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value) quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" \ f":{config_.QUICK_FLOW_POOL_ID}" self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value) return check_result """ # 本次视频都有本地记录 if len(videos) == 0: error_flag = False return check_result, error_flag # 本地无记录视频，检查实时分发数 st_remain_view_count = time.time() view_count_result, error_flag = get_videos_remain_view_count(app_type=self.app_type, videos=videos) log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'app_type': self.app_type, 'mid': self.mid, 'uid': self.uid, 'operation': 'remainViewCount', 'executeTime': (time.time() - st_remain_view_count) * 1000 }) # 判断返回的错误标记，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 = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item[0]}:{item[1]}" self.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 = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{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, flow_pool_id=None): """ 拼接key :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池} :param flow_pool_id: 流量池ID :return: key_name """ if pool_type == 'rov': # appType = 13 票圈视频app if self.app_type == config_.APP_TYPE['APP']: key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_APP # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验 # elif self.ab_code == config_.AB_CODE['rov_rank_appType_18_19']: # key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}." # 其他 else: key_name_prefix = config_.RECALL_KEY_NAME_PREFIX # 判断热度列表是否更新，未更新则使用前一天的热度列表 key_name = 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 = 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召回池数据未按时更新，请及时查看解决。'.format(config_.ENV_TEXT) # 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': if flow_pool_id == config_.QUICK_FLOW_POOL_ID: return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}" else: return f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}" elif pool_type == 'special': key_name_prefix = config_.KEY_NAME_PREFIX_SPECIAL_VIDEOS # 判断列表是否更新，未更新则使用前一天的列表 key_name = f"{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 = f"{key_name_prefix}{redis_date}" # 判断当前时间是否晚于更新时间，发送消息到飞书 # now_h = datetime.now().hour # feishu_text = '{} —— 今日special mid 数据未按时更新，请及时查看解决。'.format(config_.ENV_TEXT) # if now_h > config_.ROV_UPDATE_H: # send_msg_to_feishu(feishu_text) return key_name, redis_date else: log_.error('pool type error') return None, None def get_video_last_idx(self, video_type=''): """获取用户上一次在rov召回池对应的位置""" # if self.ab_code in [config_.AB_CODE['rank_by_h']] or self.app_type == config_.APP_TYPE['APP']: # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验 if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_h'].items()] + \ [code for _, code in config_.AB_CODE['region_rank_by_h'].items()] + \ [config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19'], config_.AB_CODE['top_video_relevant_appType_19']] + \ [code for _, code in config_.AB_CODE['rank_by_24h'].items()] or \ video_type == 'whole_movies': rov_pool_key, redis_date = self.get_pool_redis_key_with_h('rov', video_type=video_type) elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]: rov_pool_key, redis_date = self.get_pool_redis_key_with_day('dup') else: rov_pool_key, redis_date = self.get_pool_redis_key('rov') if not rov_pool_key: return None, None, None if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]: now_h = datetime.now().hour if now_h < 7: last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PRE_PREFIX else: last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_NOW_PREFIX elif video_type == 'whole_movies': last_key_prefix = config_.LAST_VIDEO_FROM_WHOLE_MOVIES_PREFIX else: last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PREFIX last_rov_recall_key = f'{last_key_prefix}{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的视频 if self.app_type == config_.APP_TYPE['APP']: key_name = config_.UPDATE_ROV_KEY_NAME_APP else: key_name = config_.UPDATE_ROV_KEY_NAME # redis_helper = RedisHelper() data = self.redis_helper.get_data_zset_with_index(key_name=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(request_id=self.request_id, 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() if self.app_type == config_.APP_TYPE['APP']: key_name = config_.TOP_VIDEO_LIST_KEY_NAME_APP else: key_name = config_.TOP_VIDEO_LIST_KEY_NAME data = self.redis_helper.get_data_from_redis(key_name=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(request_id=self.request_id, 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 [], [] def get_pool_redis_key_with_h(self, pool_type, video_type=''): """ 拼接key，获取以小时级别更新的视频列表 :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池} :param video_type: 视频列表区分 whole_movies - 完整影视资源 :return: key_name """ if pool_type == 'rov': now_date = date.today().strftime('%Y%m%d') # 获取当前所在小时 h = datetime.now().hour # appType = 13, 票圈视频APP # 数据更新周期：每天07:00-21:00, 2h/次 # if self.app_type == config_.APP_TYPE['APP']: # if h < 7: # key_h = 21 # key_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') # elif h > 21: # key_h = 21 # key_date = now_date # else: # if h % 2 == 0: # key_h = h - 1 # key_date = now_date # else: # key_h = h # key_date = now_date # # print(key_date, key_h) # key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{key_date}.{key_h}' # if self.redis_helper.key_exists(key_name): # return key_name, key_h # else: # if key_h == 7: # redis_h = 21 # redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') # else: # redis_h = key_h - 2 # redis_date = key_date # # print(redis_date, redis_h) # key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{redis_date}.{redis_h}' # return key_name, redis_h # elif self.app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]: # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验 if self.ab_code in [ config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19'], config_.AB_CODE['top_video_relevant_appType_19'] ]: # 判断热度列表是否更新，未更新则使用前一小时的热度列表 key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}:" key_name = f"{key_name_prefix}{now_date}:{h}" if self.redis_helper.key_exists(key_name): return key_name, h else: if h == 0: redis_h = 23 redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') else: redis_h = h - 1 redis_date = now_date key_name = f"{key_name_prefix}{redis_date}:{redis_h}" # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书 # now_m = datetime.now().minute # feishu_text = '{} —— appType = {}, h = {} 数据未按时更新，请及时查看解决。'.format( # config_.ENV_TEXT, self.app_type, h) # if now_m > config_.ROV_H_UPDATE_MINUTE: # send_msg_to_feishu(feishu_text) return key_name, redis_h # 完整影视资源 elif video_type == 'whole_movies': # 判断完整影视资源列表是否更新，未更新则使用前一小时的热度列表 key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_WHOLE_MOVIES}" key_name = f"{key_name_prefix}{now_date}.{h}" if self.redis_helper.key_exists(key_name): return key_name, h else: if h == 0: redis_h = 23 redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') else: redis_h = h - 1 redis_date = now_date key_name = f"{key_name_prefix}{redis_date}.{redis_h}" # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书 # now_m = datetime.now().minute # feishu_text = '{} —— appType = {}, h = {} 完整影视资源数据未按时更新，请及时查看解决。'.format( # config_.ENV_TEXT, self.app_type, h) # if now_m > config_.ROV_H_UPDATE_MINUTE: # send_msg_to_feishu(feishu_text) return key_name, redis_h else: # 判断热度列表是否更新，未更新则使用前一小时的热度列表 if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]: # 获取provinceCode province_code = self.client_info.get('provinceCode', '-1') if province_code == '': province_code = '-1' if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'): key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_24H}{province_code}." else: key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{province_code}:" elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]: key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_24H else: key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_H key_name = f"{key_prefix}{self.data_key}:{self.rule_key}:{now_date}:{h}" if self.redis_helper.key_exists(key_name): return key_name, h else: if h == 0: redis_h = 23 redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') else: redis_h = h - 1 redis_date = now_date key_name = f"{key_prefix}{self.data_key}:{self.rule_key}:{redis_date}:{redis_h}" # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书 # now_m = datetime.now().minute # feishu_text = '{} —— appType = {}, h = {}, key = {}, province_code = {} 数据未按时更新，请及时查看解决。'.format( # config_.ENV_TEXT, self.app_type, h, key_name, self.client_info.get('provinceCode', '-1')) # if now_m > config_.ROV_H_UPDATE_MINUTE: # send_msg_to_feishu(feishu_text) return key_name, redis_h 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 flow_pool_recall_18_19(self, size=4, push_from=config_.PUSH_FROM['flow_recall']): """从流量池中获取视频""" start_time = time.time() flow_pool_key = self.get_pool_redis_key('flow') flow_pool_recall_result = [] flow_pool_recall_videos = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 idx = 0 while len(flow_pool_recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_FLOW_POOL_18_19: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key, start=idx, end=idx + get_size - 1, with_scores=True) # 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 与 score做mapping整理 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(request_id=self.request_id, 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() # 添加视频源参数 pushFrom, abCode if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] flow_pool_recall_result.extend(temp_result) idx += get_size log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'flow_pool_recall_18_19', 'executeTime': (time.time() - start_time) * 1000 }) return flow_pool_recall_result[:size] def get_pool_redis_key_with_day(self, pool_type): """ 拼接key，获取以天级别规则更新的视频列表 :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池} :return: key_name """ now_date = datetime.today() now_dt = datetime.strftime(now_date, '%Y%m%d') if pool_type == 'rov': # 判断列表是否更新，未更新则使用前一天的列表 rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{now_dt}' if self.redis_helper.key_exists(key_name=rule_key_name): return rule_key_name, now_dt else: redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d') rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{redis_dt}' # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书 # now_h = datetime.now().hour # feishu_text = f'{config_.ENV_TEXT} —— appType = {self.app_type}, date = {now_dt}, ' \ # f'rule_key = {self.rule_key} 数据未按时更新，请及时查看解决。' # if now_h > config_.ROV_DAY_UPDATE_MINUTE: # send_msg_to_feishu(feishu_text) return rule_key_name, redis_dt elif pool_type == 'dup': now_h = datetime.now().hour if now_h < 7: dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE else: dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_NOW dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{now_dt}' if self.redis_helper.key_exists(key_name=dup_key_name): return dup_key_name, now_dt else: redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d') dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{redis_dt}' return dup_key_name, redis_dt 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_day(self): """获取用户上一次在天级规则更新列表中对应的位置""" rule_key_name, redis_dt = self.get_pool_redis_key_with_day('rov') if not rule_key_name: return None, None, None last_rule_day_recall_key = \ f'{config_.LAST_VIDEO_FROM_RULE_DAY_POOL_PREFIX}{self.app_type}.{self.mid}.{redis_dt}' value = self.redis_helper.get_data_from_redis(last_rule_day_recall_key) if value: idx = self.redis_helper.get_index_with_data(rule_key_name, value) if not idx: idx = 0 else: idx += 1 else: idx = 0 return rule_key_name, last_rule_day_recall_key, idx def old_videos_recall(self, size): """老视频召回""" # 获取老视频 now_dt = datetime.strftime(datetime.today(), '%Y%m%d') key_name = f'{config_.RECALL_KEY_NAME_PREFIX_OLD_VIDEOS}{now_dt}' old_videos = self.redis_helper.get_data_from_set(key_name=key_name) if not old_videos: return [] # 过滤 old_video_ids = [int(video_id) for video_id in old_videos] filter_ = FilterVideos(request_id=self.request_id, app_type=self.app_type, video_ids=old_video_ids, mid=self.mid, uid=self.uid) ge = gevent.spawn(filter_.filter_videos) ge.join() filtered_videos = ge.get() if not filtered_videos: return [] # 添加视频源参数 pushFrom, abCode old_video_result = [{'videoId': int(item), 'rovScore': 0, 'pushFrom': config_.PUSH_FROM['old_video'], 'abCode': self.ab_code} for item in filtered_videos] # 随机抽取 size+1 条数据 random.shuffle(old_video_result) return old_video_result[:size+1] def get_region_dup_video_last_idx_h(self, province_code, region_dup=None): """获取用户上一次在 地域分组 相关去重列表中对应的位置""" now_date = date.today().strftime('%Y%m%d') h = datetime.now().hour if region_dup == 1: # 小程序地域分组天级更新结果与小程序地域分组小时级更新结果去重 key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_DAY_H elif region_dup == 2: if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'): # 小程序天级更新结果与 小程序地域分组小时级更新24h结果 去重 key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_DAY_24H else: # 小程序天级更新结果与 小程序地域分组天级更新结果/小程序地域分组小时级更新结果 去重 key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_DAY_H else: key_name_prefix = '' key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{now_date}.{h}" last_region_dup_key = \ f'{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{h}' if not self.redis_helper.key_exists(key_name=key_name): if h == 0: redis_h = 23 redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d') else: redis_h = h - 1 redis_date = now_date key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{redis_date}.{redis_h}" last_region_dup_key = \ f"{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{redis_h}" # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书 # now_m = datetime.now().minute # feishu_text = f"{config_.ENV_TEXT} —— appType = {self.app_type}, h = {h}, key = {key_name}, " \ # f"province_code = {province_code} 数据未按时更新，请及时查看解决。" # if now_m > config_.REGION_H_UPDATE_MINUTE: # send_msg_to_feishu(feishu_text) value = self.redis_helper.get_data_from_redis(last_region_dup_key) if value: idx = self.redis_helper.get_index_with_data(key_name, value) if not idx: idx = 0 else: idx += 1 else: idx = 0 return key_name, last_region_dup_key, idx def rov_pool_recall_with_region(self, size=4, expire_time=24*3600): """ 地域分组召回视频 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :return: """ start_time = time.time() # 获取provinceCode province_code = self.client_info.get('provinceCode', '-1') if province_code == '': province_code = '-1' # if self.ab_code in [config_.AB_CODE['region_rank_by_h'].get('region_rule_rank5_appType_0_data1'), # config_.AB_CODE['region_rank_by_h'].get('abtest_082'), # config_.AB_CODE['region_rank_by_h'].get('abtest_112')]: if province_code == '-1': t = [ # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'), gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup2', expire_time), gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup3', expire_time)] else: t = [ # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size), # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'region_24h'), # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'), gevent.spawn(self.recall_region_dup_24h, province_code, size, 'region_h', expire_time), gevent.spawn(self.recall_region_dup_24h, province_code, size, 'region_24h', expire_time), gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup2', expire_time), gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup3', expire_time)] # else: # if province_code == '-1': # # t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h')] # t = [gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup2', expire_time)] # # else: # t = [ # # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size), # # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'region_24h'), # # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'), # gevent.spawn(self.recall_region_dup_24h, province_code, size, 'region_h', expire_time), # gevent.spawn(self.recall_region_dup_24h, province_code, size, 'region_24h', expire_time), # gevent.spawn(self.recall_region_dup_24h, province_code, size, '24h_dup2', expire_time), # # # gevent.spawn(self.region_dup_recall, province_code, 1, size, expire_time), # # gevent.spawn(self.region_dup_recall, province_code, 2, size, expire_time) # ] gevent.joinall(t) region_recall_result_list = [i.get() for i in t] # 将已获取到的视频按顺序去重合并 now_video_ids = [] recall_result = [] for region_result in region_recall_result_list: for video in region_result: video_id = video.get('videoId') if video_id not in now_video_ids: recall_result.append(video) now_video_ids.append(video_id) if len(recall_result) >= size: break else: continue # 判断获取到的小时级数据数量 if len(recall_result) < size: # 补充数据 rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) # 去重合并 for video in rov_recall_result: vid = video.get('videoId') if vid not in now_video_ids: recall_result.append(video) now_video_ids.append(vid) if len(recall_result) >= size: break else: continue log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'rov_pool_recall_with_region', 'executeTime': (time.time() - start_time) * 1000 }) return recall_result[:size] def rov_pool_recall_with_region_by_h(self, province_code, size=4, key_flag=''): """ 地域分组小时级视频召回 :param size: 视频数 :param province_code: 省份code :param key_flag: :return: """ if key_flag == 'region_24h': push_from = config_.PUSH_FROM['rov_recall_region_24h'] elif key_flag == 'day_24h': push_from = config_.PUSH_FROM['rov_recall_24h'] else: push_from = config_.PUSH_FROM['rov_recall_region_h'] # 获取mid对应的小时级列表redis-key h_recall_mid_key = self.get_mid_h_key(province_code=province_code, key_flag=key_flag) if not self.redis_helper.key_exists(h_recall_mid_key): recall_result = [] else: # 过滤的视频 fil_video_ids = [] recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key, start=(freq - 1) * get_size, end=freq * get_size - 1, with_scores=True) if not data: # log_.info('地域分组小时级更新视频已取完') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.data_key, self.ab_code, province_code, key_flag) ge.join() filtered_result = ge.get() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] recall_result.extend(temp_result) fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result]))) else: fil_video_ids.extend(video_ids) # 将被过滤的视频进行移除 for value in fil_video_ids: self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value) return recall_result[:size] def region_dup_recall(self, province_code, region_dup, size=4, expire_time=23*3600): """ region dup 更新列表视频召回 :param province_code: :param region_dup: :param size: :param expire_time: :return: """ if region_dup == 1: push_from = config_.PUSH_FROM['rov_recall_region_day'] elif region_dup == 2: push_from = config_.PUSH_FROM['rov_recall_day'] # 获取region dup更新列表相关redis key, 用户上一次在列表对应的位置 key_name, last_region_dup_key, idx = self.get_region_dup_video_last_idx_h( province_code=province_code, region_dup=region_dup) # 获取天级规则更新列表数据 if not key_name: # log_.info(f'region dup 更新列表中无视频, region_dup = {region_dup}') recall_result = [] else: recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=key_name, start=idx, end=idx + get_size - 1, with_scores=True) if not data: # log_.info(f'region dup 更新视频已取完, region_dup = {region_dup}') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, 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() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] recall_result.extend(temp_result) else: # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天 if self.mid: # mid为空时，不做记录 self.redis_helper.set_data_to_redis(key_name=last_region_dup_key, value=data[-1][0], expire_time=expire_time) idx += get_size return recall_result[:size] def rule_recall_by_h(self, size=4, expire_time=24*3600): """ 小时级召回视频 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :return: """ start_time = time.time() t = [gevent.spawn(self.rov_pool_recall_h, size), gevent.spawn(self.rov_pool_recall_h, size, '24h')] gevent.joinall(t) h_recall_result_list = [i.get() for i in t] # 将已获取到的视频按顺序去重合并 now_video_ids = [] recall_result = [] for h_result in h_recall_result_list: for video in h_result: video_id = video.get('videoId') if video_id not in now_video_ids: recall_result.append(video) now_video_ids.append(video_id) if len(recall_result) >= size: break else: continue # 判断获取到的小时级数据数量 if len(recall_result) < size: # 补充数据 rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) # 去重合并 for video in rov_recall_result: vid = video.get('videoId') if vid not in now_video_ids: recall_result.append(video) now_video_ids.append(vid) if len(recall_result) >= size: break else: continue log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': 'rule_recall_by_h', 'executeTime': (time.time() - start_time) * 1000 }) return recall_result[:size] def rov_pool_recall_h(self, size=4, key_flag=''): """ 小时级视频召回 :param size: 视频数 :param key_flag: :return: """ if key_flag == '24h': push_from = config_.PUSH_FROM['rov_recall_24h'] else: push_from = config_.PUSH_FROM['rov_recall_h'] # 获取mid对应的小时级列表redis-key h_recall_mid_key = self.get_mid_h_key(province_code='', key_flag=key_flag) if not self.redis_helper.key_exists(h_recall_mid_key): recall_result = [] else: # 过滤的视频 fil_video_ids = [] recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key, start=(freq - 1) * get_size, end=freq * get_size - 1, with_scores=True) if not data: # log_.info('小时级更新视频已取完') break # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids) ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.ab_code, '', key_flag) ge.join() filtered_result = ge.get() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] recall_result.extend(temp_result) fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result]))) else: fil_video_ids.extend(video_ids) # 将被过滤的视频进行移除 for value in fil_video_ids: self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value) return recall_result[:size] def get_relevant_videos_19(self, video_id, size=4): """ 获取最惊奇相关推荐视频 :param video_id: 头部视频id :return: relevant_result """ push_from = config_.PUSH_FROM['top_video_relevant_appType_19'] relevant_result = [] relevant_videos_key_name = f"{config_.MOVIE_RELEVANT_LIST_KEY_NAME_PREFIX}{video_id}" # redis_helper = RedisHelper() if not self.redis_helper.key_exists(key_name=relevant_videos_key_name): return relevant_result # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=relevant_videos_key_name, start=0, end=-1, with_scores=True) if not data: return relevant_result # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, 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() if filtered_result: # 添加视频源参数 pushFrom, abCode relevant_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] return relevant_result[:size] def relevant_recall_19(self, video_id, size=4, expire_time=24*3600): """ 最惊奇相关推荐视频召回 :param video_id: :param size: :param expire_time: :return: """ t = [gevent.spawn(self.get_relevant_videos_19, video_id, size), gevent.spawn(self.rov_pool_recall, size, expire_time)] gevent.joinall(t) relevant_recall_result_list = [i.get() for i in t] # 将已获取到的视频按顺序去重合并 now_video_ids = [] recall_result = [] for relevant_result in relevant_recall_result_list: for video in relevant_result: video_id = video.get('videoId') if video_id not in now_video_ids: recall_result.append(video) now_video_ids.append(video_id) return recall_result[:size] def rov_pool_recall_19(self, size=4, expire_time=24*3600): """ 最惊奇视频召回 :param size: 获取视频个数 :param expire_time: 末位视频记录redis过期时间 :return: """ t = [gevent.spawn(self.rov_pool_recall, size, expire_time=3600, video_type='whole_movies', push_from=config_.PUSH_FROM['whole_movies']), gevent.spawn(self.flow_pool_recall_18_19, size, push_from=config_.PUSH_FROM['talk_videos'])] gevent.joinall(t) recall_result_list = [i.get() for i in t] # 将已获取到的视频按顺序去重合并 now_video_ids = [] recall_result = [] for item in recall_result_list: for video in item: video_id = video.get('videoId') if video_id not in now_video_ids: recall_result.append(video) now_video_ids.append(video_id) if len(recall_result) >= size: break else: continue # 判断获取到的小时级数据数量 if len(recall_result) < size: # 补充数据 rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time) # 去重合并 for video in rov_recall_result: vid = video.get('videoId') if vid not in now_video_ids: recall_result.append(video) now_video_ids.append(vid) if len(recall_result) >= size: break else: continue return recall_result[:size] def update_last_video_record(self, record_key, pool_key_prefix, province_code): # 判断当前小时的小时级列表是否更新 now_date = datetime.today() h = datetime.now().hour now_dt = datetime.strftime(now_date, '%Y%m%d') now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}" if self.redis_helper.key_exists(key_name=now_pool_recall_key): value = {'date': now_dt, 'h': h} self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600) else: if h == 0: redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d') redis_h = 23 else: redis_dt = now_dt redis_h = h - 1 now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{redis_dt}:{redis_h}" value = {'date': redis_dt, 'h': redis_h} self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600) return now_pool_recall_key def get_video_idx(self, pool_recall_key, last_video_key): """ 获取上次视频所在位置 :param pool_recall_key: 视频所在列表 key :param last_video_key: 上一次记录的视频 :return: idx """ value = self.redis_helper.get_data_from_redis(last_video_key) if value: idx = self.redis_helper.get_index_with_data(key_name=pool_recall_key, value=value) if not idx: idx = 0 else: idx += 1 else: idx = 0 return idx def get_last_recommend_video_idx(self, province_code, record_key_prefix, pool_key_prefix, last_video_key_prefix): # 判断mid对应上一次视频位置 时间记录 record_key = f"{record_key_prefix}{self.app_type}:{self.mid}" last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}' if not self.redis_helper.key_exists(key_name=record_key): # ###### 记录key不存在 self.redis_helper.del_keys(key_name=last_video_key) idx = 0 pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix, province_code=province_code) else: # ###### 记录key存在，判断date, h now_date = datetime.today() h = datetime.now().hour # 获取记录的date, h record = self.redis_helper.get_data_from_redis(key_name=record_key) record_dt = eval(record).get('date') record_h = eval(record).get('h') now_dt = datetime.strftime(now_date, '%Y%m%d') if record_dt == now_dt and int(record_h) == h: # 已获取当前小时数据 pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}" idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key) elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23): # 记录的h - 当前h = 1，判断当前h数据是否已更新 now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}" if self.redis_helper.key_exists(key_name=now_pool_recall_key): new_record = {'date': now_dt, 'h': h} self.redis_helper.set_data_to_redis(key_name=record_key, value=str(new_record), expire_time=2*3600) idx = 0 self.redis_helper.del_keys(key_name=last_video_key) pool_recall_key = now_pool_recall_key else: pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{record_dt}:{record_h}" idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key) else: idx = 0 self.redis_helper.del_keys(key_name=last_video_key) pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix, province_code=province_code) return pool_recall_key, idx def recall_region_dup_24h(self, province_code, size=4, key_flag='', expire_time=2*3600): """ 从小程序小时级24h数据 筛选后的剩余数据 更新结果中获取视频 :param size: 获取视频个数 :param key_flag: 视频表标记 :param expire_time: 末位视频记录redis过期时间 :return: """ start_time = time.time() if key_flag == 'region_h': # 分地域相对24h的筛选结果 # mid对应上一次视频位置 时间记录 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_H # 视频列表 pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H # mid对应上一次视频记录 last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_H_PREFIX push_from = config_.PUSH_FROM['rov_recall_region_h'] elif key_flag == 'region_24h': # 分地域相对24h的筛选结果 # mid对应上一次视频位置 时间记录 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP1_24H # 视频列表 pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H # mid对应上一次视频记录 last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP1_24H_PREFIX push_from = config_.PUSH_FROM['rov_recall_region_24h'] elif key_flag == '24h_dup2': # 不分地域相对24h的筛选结果 # mid对应上一次视频位置 时间记录 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP2_24H # 视频列表 pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H # mid对应上一次视频记录 last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP2_24H_PREFIX push_from = config_.PUSH_FROM['rov_recall_24h'] elif key_flag == '24h_dup3': # 不分地域相对24h的筛选后剩余数据 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP3_24H pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP3_24H_PREFIX push_from = config_.PUSH_FROM['rov_recall_24h_dup'] elif key_flag == '48h_dup2': # 不分地域相对48h的筛选结果 # mid对应上一次视频位置 时间记录 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP2_48H # 视频列表 pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_48H_H # mid对应上一次视频记录 last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP2_48H_PREFIX push_from = config_.PUSH_FROM['rov_recall_48h'] elif key_flag == '48h_dup3': # 不分地域相对48h的筛选后剩余数据 record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP3_48H pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_48H_H last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP3_48H_PREFIX push_from = config_.PUSH_FROM['rov_recall_48h_dup'] else: return [] # 获取相关redis key, 用户上一次在rov召回池对应的位置 pool_key, idx = self.get_last_recommend_video_idx(province_code=province_code, record_key_prefix=record_key_prefix, pool_key_prefix=pool_key_prefix, last_video_key_prefix=last_video_key_prefix) if not pool_key: return [] recall_data = [] pool_recall_result = [] # 每次获取的视频数 get_size = size * 5 # 记录获取频次 freq = 0 while len(pool_recall_result) < size: freq += 1 if freq > config_.MAX_FREQ_FROM_ROV_POOL: break # 获取数据 data = self.redis_helper.get_data_zset_with_index(key_name=pool_key, start=idx, end=idx + get_size - 1, with_scores=True) if not data: break recall_data.extend(data) # 获取视频id,并转换类型为int,并存储为key-value{videoId: score} 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(request_id=self.request_id, 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() if filtered_result: # 添加视频源参数 pushFrom, abCode temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)], 'pushFrom': push_from, 'abCode': self.ab_code} for item in filtered_result if video_score.get(int(item)) is not None] pool_recall_result.extend(temp_result) else: # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天 if self.mid and self.ab_code != config_.AB_CODE['region_rank_by_h'].get('abtest_112'): # mid为空时，不做记录 last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}' self.redis_helper.set_data_to_redis(key_name=last_video_key, value=data[-1][0], expire_time=expire_time) idx += get_size pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True) if len(recall_data) > 0 and len(pool_recall_result) == 0 \ and self.ab_code == config_.AB_CODE['region_rank_by_h'].get('abtest_112') and self.mid: # 召回数据不为空 & 过滤后结果为空 & 位于实验组 & mid不为空时，更新召回获取的末位视频id记录到定位的key中 last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}' self.redis_helper.set_data_to_redis(key_name=last_video_key, value=recall_data[-1][0], expire_time=expire_time) log_.info({ 'logTimestamp': int(time.time() * 1000), 'request_id': self.request_id, 'operation': push_from, 'pool_recall_result': pool_recall_result, 'executeTime': (time.time() - start_time) * 1000 }) return pool_recall_result[:size]