import datetime
import pandas as pd
import math
from odps import ODPS
from threading import Timer
from get_data import get_data_from_odps
from utils import filter_video_status
from db_helper import RedisHelper
from config import set_config
from log import Log
config_, _ = set_config()
log_ = Log()
project = 'loghubods'
table = 'video_each_hour_update'
features = [
'videoid',
'lastonehour_preview', # 过去1小时预曝光人数
'lastonehour_view', # 过去1小时曝光人数
'lastonehour_play', # 过去1小时播放人数
'lastonehour_share', # 过去1小时分享人数
'lastonehour_return', # 过去1小时分享，过去1小时回流人数
'lastonehour_preview_total_final', # 过去1小时预曝光次数
'lastonehour_view_total_final', # 过去1小时曝光次数
'lastonehour_play_total_final', # 过去1小时播放次数
'lastonehour_share_total_final', # 过去1小时分享次数
'lastonehour_show', # 过去1小时video_show人数
'lastonehour_show_total_final', # 过去1小时video_show次数
'platform_return',
]
def h_data_check(project, table, now_date):
"""检查数据是否准备好"""
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:
dt = datetime.datetime.strftime(now_date, '%Y%m%d%H')
sql = f'select * from {project}.{table} where dt = {dt}'
with odps.execute_sql(sql=sql).open_reader() as reader:
data_count = reader.count
except Exception as e:
data_count = 0
return data_count
def get_rov_redis_key(now_date):
# 获取rov模型结果存放key
redis_helper = RedisHelper()
now_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{now_dt}'
if not redis_helper.key_exists(key_name=key_name):
pre_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
return key_name
def get_feature_data(now_date):
"""获取特征数据"""
dt = datetime.datetime.strftime(now_date, '%Y%m%d%H')
# dt = '2022041310'
records = get_data_from_odps(date=dt, project=project, table=table)
feature_data = []
for record in records:
item = {}
for feature_name in features:
item[feature_name] = record[feature_name]
feature_data.append(item)
feature_df = pd.DataFrame(feature_data)
return feature_df
def cal_score(df, param):
"""
计算score
:param df: 特征数据
:param param: 规则参数
:return:
"""
# score计算公式: sharerate*backrate*logback*ctr
# sharerate = lastonehour_share/(lastonehour_play+1000)
# backrate = lastonehour_return/(lastonehour_share+10)
# ctr = lastonehour_play/(lastonehour_view+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
# score = sharerate * backrate * LOG(lastonehour_return+1) * K2
df = df.fillna(0)
df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000)
df['back_rate'] = df['lastonehour_return'] / (df['lastonehour_share'] + 10)
df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
if param.get('view_type', None) == 'pre-view':
df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
elif param.get('view_type', None) == 'video-show':
df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
else:
df['ctr'] = df['lastonehour_play'] / (df['lastonehour_view'] + 1000)
df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
df = df.sort_values(by=['score'], ascending=False)
return df
def cal_score2(df):
# score2计算公式: score = lastonehour_return/(lastonehour_view+1000)
df = df.fillna(0)
df['score'] = df['lastonehour_return'] / (df['lastonehour_view'] + 1000)
df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
df = df.sort_values(by=['score'], ascending=False)
return df
def cal_score3(df):
# score3计算公式:
# score = lastonehour_share_total_final/(lastonehour_view+1000)
# + 0.03 * lastonehour_return/(lastonehour_share_total_final+1)
df = df.fillna(0)
df['share_rate'] = df['lastonehour_share_total_final'] / (df['lastonehour_view'] + 1000)
df['back_rate'] = df['lastonehour_return'] / (df['lastonehour_share_total_final'] + 1)
df['score'] = df['share_rate'] + 0.03 * df['back_rate']
df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
df = df.sort_values(by=['score'], ascending=False)
return df
def video_rank(df, now_date, now_h, rule_key, param):
"""
获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
:param df:
:param now_date:
:param now_h:
:param rule_key: 小时级数据进入条件
:param param: 小时级数据进入条件参数
:return:
"""
# 获取rov模型结果
redis_helper = RedisHelper()
key_name = get_rov_redis_key(now_date=now_date)
initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
log_.info(f'initial data count = {len(initial_data)}')
# 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
return_count = param.get('return_count')
score_value = param.get('score_rule')
platform_return_rate = param.get('platform_return_rate', 0)
h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= score_value)
& (df['platform_return_rate'] >= platform_return_rate)]
h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
h_recall_videos = h_recall_df['videoid'].to_list()
log_.info(f'h_recall videos count = {len(h_recall_videos)}')
# 视频状态过滤
filtered_videos = filter_video_status(h_recall_videos)
log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
# 写入对应的redis
h_video_ids = []
h_recall_result = {}
for video_id in filtered_videos:
score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
h_recall_result[int(video_id)] = float(score)
h_video_ids.append(int(video_id))
h_recall_key_name = \
f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
if len(h_recall_result) > 0:
redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=23 * 3600)
# 清空线上过滤应用列表
redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER}{rule_key}")
dup_to_redis(h_video_ids, now_date, now_h, rule_key)
# 去重更新rov模型结果，并另存为redis中
# initial_data_dup = {}
# for video_id, score in initial_data:
# if int(video_id) not in h_video_ids:
# initial_data_dup[int(video_id)] = score
# log_.info(f"initial data dup count = {len(initial_data_dup)}")
# initial_key_name = \
# f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H}{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
# if len(initial_data_dup) > 0:
# redis_helper.add_data_with_zset(key_name=initial_key_name, data=initial_data_dup, expire_time=23 * 3600)
def dup_to_redis(h_video_ids, now_date, now_h, rule_key):
"""将小时级数据与其他召回视频池去重，存入对应的redis"""
redis_helper = RedisHelper()
# ##### 去重小程序相对24h数据更新结果，并另存为redis中
rule_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}rule2.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
if redis_helper.key_exists(key_name=rule_24h_key_name):
rule_24h_data = redis_helper.get_all_data_from_zset(key_name=rule_24h_key_name, with_scores=True)
log_.info(f'rule_24h data count = {len(rule_24h_data)}')
rule_24h_dup = {}
for video_id, score in rule_24h_data:
if int(video_id) not in h_video_ids:
rule_24h_dup[int(video_id)] = score
h_video_ids.append(int(video_id))
log_.info(f"rule_24h data dup count = {len(rule_24h_dup)}")
rule_24h_dup_key_name = \
f"{config_.RECALL_KEY_NAME_PREFIX_DUP_24H_H}{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
if len(rule_24h_dup) > 0:
redis_helper.add_data_with_zset(key_name=rule_24h_dup_key_name, data=rule_24h_dup, expire_time=23 * 3600)
# ##### 去重小程序模型更新结果，并另存为redis中
model_key_name = get_rov_redis_key(now_date=now_date)
model_data = redis_helper.get_all_data_from_zset(key_name=model_key_name, with_scores=True)
log_.info(f'model data count = {len(model_data)}')
model_data_dup = {}
for video_id, score in model_data:
if int(video_id) not in h_video_ids:
model_data_dup[int(video_id)] = score
h_video_ids.append(int(video_id))
log_.info(f"model data dup count = {len(model_data_dup)}")
model_data_dup_key_name = \
f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H}{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
if len(model_data_dup) > 0:
redis_helper.add_data_with_zset(key_name=model_data_dup_key_name, data=model_data_dup, expire_time=23 * 3600)
def rank_by_h(now_date, now_h, rule_params):
# 获取特征数据
feature_df = get_feature_data(now_date=now_date)
# rank
for key, value in rule_params.items():
log_.info(f"rule = {key}, param = {value}")
# 计算score
cal_score_func = value.get('cal_score_func', 0)
if cal_score_func == 2:
score_df = cal_score2(df=feature_df)
elif cal_score_func == 3:
score_df = cal_score3(df=feature_df)
else:
score_df = cal_score(df=feature_df, param=value)
video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value)
# to-csv
score_filename = f"score_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
score_df.to_csv(f'./data/{score_filename}')
# to-logs
log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d%H'),
"redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_BY_H,
"rule_key": key,
"score_df": score_df[['videoid', 'score']]})
def h_rank_bottom(now_date, now_h, rule_key):
"""未按时更新数据，用上一小时结果作为当前小时的数据"""
log_.info(f"rule_key = {rule_key}")
# 获取rov模型结果
redis_helper = RedisHelper()
if now_h == 0:
redis_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
redis_h = 23
else:
redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
redis_h = now_h - 1
key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_H]
for key_prefix in key_prefix_list:
key_name = f"{key_prefix}{rule_key}.{redis_dt}.{redis_h}"
initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
if initial_data is None:
initial_data = []
final_data = dict()
h_video_ids = []
for video_id, score in initial_data:
final_data[video_id] = score
h_video_ids.append(int(video_id))
# 存入对应的redis
final_key_name = \
f"{key_prefix}{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
if len(final_data) > 0:
redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=23 * 3600)
# 清空线上过滤应用列表
redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER}{rule_key}")
dup_to_redis(h_video_ids, now_date, now_h, rule_key)
def h_timer_check():
rule_params = config_.RULE_PARAMS
# return_count_list = [20, 10]
now_date = datetime.datetime.today()
log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
now_h = datetime.datetime.now().hour
now_min = datetime.datetime.now().minute
# if now_h == 0:
# for key, _ in rule_params.items():
# h_rank_bottom(now_date=now_date, now_h=now_h, rule_key=key)
# return
# 查看当前小时更新的数据是否已准备好
h_data_count = h_data_check(project=project, table=table, now_date=now_date)
if h_data_count > 0:
log_.info(f'h_data_count = {h_data_count}')
# 数据准备好，进行更新
rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params)
elif now_min > 50:
log_.info('h_recall data is None, use bottom data!')
for key, _ in rule_params.items():
h_rank_bottom(now_date=now_date, now_h=now_h, rule_key=key)
else:
# 数据没准备好，1分钟后重新检查
Timer(60, h_timer_check).start()
if __name__ == '__main__':
# df1 = get_feature_data()
# res = cal_score(df=df1)
# video_rank(df=res, now_date=datetime.datetime.today())
# rank_by_h()
h_timer_check()