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']:
# 获取当前所在小时
h = datetime.now().hour
# 判断热度列表是否更新，未更新则使用前一小时的热度列表
key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, h)
if self.redis_helper.key_exists(key_name):
return key_name, h
else:
if h == 0:
redis_h = 23
else:
redis_h = h - 1
key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, redis_h)
# 判断当前时间是否晚于数据正常更新时间，发送消息到飞书
now_m = datetime.now().minute
feishu_text = '{} —— appType = {}, h = {} 数据未按时更新，请及时查看解决。'.format(
config_.ENV_TEXT, self.app_type, h)
if now_m > config_.ROV_UPDATE_MINUTE_6:
send_msg_to_feishu(feishu_text)
return key_name, redis_h
# 其他
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召回池数据未按时更新，请及时查看解决。'.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':
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 [], []