import pandas as pd
import math
import traceback
from functools import reduce
from odps import ODPS
from threading import Timer
from datetime import datetime, timedelta
from get_data import get_data_from_odps
from db_helper import RedisHelper
from utils import filter_video_status, check_table_partition_exits, filter_video_status_app, \
request_post, send_msg_to_feishu
from config import set_config
from log import Log
config_, _ = set_config()
log_ = Log()
features = [
'apptype',
'videoid',
'preview人数', # 过去24h预曝光人数
'view人数', # 过去24h曝光人数
'play人数', # 过去24h播放人数
'share人数', # 过去24h分享人数
'回流人数', # 过去24h分享，过去24h回流人数
'preview次数', # 过去24h预曝光次数
'view次数', # 过去24h曝光次数
'play次数', # 过去24h播放次数
'share次数', # 过去24h分享次数
'platform_return',
'platform_preview',
'platform_preview_total',
'platform_show',
'platform_show_total',
'platform_view',
'platform_view_total',
]
def get_rov_redis_key(now_date):
# 获取rov模型结果存放key
redis_helper = RedisHelper()
now_dt = 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.strftime(now_date - timedelta(days=1), '%Y%m%d')
key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
return key_name
def h_data_check(project, table, now_date, now_h):
"""检查数据是否准备好"""
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:
# 23点开始到8点之前（不含8点），全部用22点生成那个列表
if now_h == 23:
dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
elif now_h < 8:
dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
else:
dt = datetime.strftime(now_date, '%Y%m%d%H')
check_res = check_table_partition_exits(date=dt, project=project, table=table)
if check_res:
sql = f'select * from {project}.{table} where dt = {dt}'
with odps.execute_sql(sql=sql).open_reader() as reader:
data_count = reader.count
else:
data_count = 0
except Exception as e:
data_count = 0
return data_count
def get_feature_data(now_date, now_h, project, table):
"""获取特征数据"""
# 23点开始到8点之前（不含8点），全部用22点生成那个列表
if now_h == 23:
dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
elif now_h < 8:
dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
else:
dt = datetime.strftime(now_date, '%Y%m%d%H')
log_.info({'feature_dt': dt})
# dt = '20220425'
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_score1(df):
# score1计算公式: score = 回流人数/(view人数+10000)
df = df.fillna(0)
df['score'] = df['回流人数'] / (df['view人数'] + 1000)
df = df.sort_values(by=['score'], ascending=False)
return df
def cal_score2(df, param):
# score2计算公式: score = share次数/(view+1000)+0.01*return/(share次数+100)
df = df.fillna(0)
if param.get('view_type', None) == 'video-show':
df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
elif param.get('view_type', None) == 'preview':
df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
else:
df['share_rate'] = df['share次数'] / (df['view人数'] + 1000)
df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
df['score'] = df['share_rate'] + 0.01 * df['back_rate']
df['platform_return_rate'] = df['platform_return'] / df['回流人数']
df = df.sort_values(by=['score'], ascending=False)
return df
def cal_score(df, param):
# score计算公式: score1 = share次数/(view+1000)+0.01*return/(share次数+100)
# ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
# score = 0.3 * score1 + 0.7 * K2
df = df.fillna(0)
if param.get('view_type', None) == 'video-show':
df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)
elif param.get('view_type', None) == 'preview':
df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
df['ctr'] = df['play人数'] / (df['preview人数'] + 1000)
else:
df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)
df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
df['platform_return_rate'] = df['platform_return'] / df['回流人数']
df['score1'] = df['share_rate'] + 0.01 * df['back_rate']
click_score_rate = param.get('click_score_rate', None)
back_score_rate = param.get('click_score_rate', None)
if click_score_rate is not None:
df['score'] = (1 - click_score_rate) * df['score1'] + click_score_rate * df['K2']
elif back_score_rate is not None:
df['score'] = (1 - back_score_rate) * df['score1'] + back_score_rate * df['back_rate']
else:
df['score'] = df['score1']
df = df.sort_values(by=['score'], ascending=False)
return df
def video_rank_h(df, now_date, now_h, rule_key, param, data_key, notify_backend):
"""
获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
:param df:
:param now_date:
:param now_h:
:param rule_key: 天级规则数据进入条件
:param param: 天级规则数据进入条件参数
:param data_key: 使用数据标识
:param notify_backend: 是否同步给后端标识
:return:
"""
redis_helper = RedisHelper()
log_.info(f"videos_count = {len(df)}")
# videoid重复时，保留分值高
df = df.sort_values(by=['score'], ascending=False)
df = df.drop_duplicates(subset=['videoid'], keep='first')
df['videoid'] = df['videoid'].astype(int)
# 获取符合进入召回源条件的视频
return_count = param.get('return_count')
if return_count:
day_recall_df = df[df['回流人数'] > return_count]
else:
day_recall_df = df
platform_return_rate = param.get('platform_return_rate', 0)
day_recall_df = day_recall_df[day_recall_df['platform_return_rate'] > platform_return_rate]
day_recall_videos = day_recall_df['videoid'].to_list()
log_.info(f'h_by24h_recall videos count = {len(day_recall_videos)}')
# 视频状态过滤
if data_key in ['data7', ]:
filtered_videos = filter_video_status_app(day_recall_videos)
else:
filtered_videos = filter_video_status(day_recall_videos)
log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
# 写入对应的redis
now_dt = datetime.strftime(now_date, '%Y%m%d')
day_video_ids = []
day_recall_result = {}
# json_data = []
for video_id in filtered_videos:
score = day_recall_df[day_recall_df['videoid'] == video_id]['score']
day_recall_result[int(video_id)] = float(score)
day_video_ids.append(int(video_id))
# json_data.append({'videoId': video_id, 'rovScore': float(score)})
h_24h_recall_key_name = \
f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{rule_key}:{now_dt}:{now_h}"
if len(day_recall_result) > 0:
log_.info(f"count = {len(day_recall_result)}, key = {h_24h_recall_key_name}")
redis_helper.add_data_with_zset(key_name=h_24h_recall_key_name, data=day_recall_result, expire_time=2 * 3600)
# 清空线上过滤应用列表
# redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")
# 去重筛选结果，保留剩余数据并写入Redis
all_videos = df['videoid'].to_list()
log_.info(f'h_by24h_recall all videos count = {len(all_videos)}')
# 视频状态过滤
if data_key in ['data7', ]:
all_filtered_videos = filter_video_status_app(all_videos)
else:
all_filtered_videos = filter_video_status(all_videos)
log_.info(f'all_filtered_videos count = {len(all_filtered_videos)}')
# 与筛选结果去重
other_videos = [video for video in all_filtered_videos if video not in day_video_ids]
log_.info(f'other_videos count = {len(other_videos)}')
# 写入对应的redis
other_24h_recall_result = {}
json_data = []
for video_id in other_videos:
score = df[df['videoid'] == video_id]['score']
other_24h_recall_result[int(video_id)] = float(score)
json_data.append({'videoId': video_id, 'rovScore': float(score)})
# other_h_24h_recall_key_name = \
# f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{app_type}:{data_key}:{rule_key}:{now_dt}:{now_h}"
other_h_24h_recall_key_name = \
f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:{rule_key}:{now_dt}:{now_h}"
if len(other_24h_recall_result) > 0:
log_.info(f"count = {len(other_24h_recall_result)}")
redis_helper.add_data_with_zset(key_name=other_h_24h_recall_key_name, data=other_24h_recall_result,
expire_time=2 * 3600)
# 通知后端更新兜底视频数据
if notify_backend is True:
log_.info('json_data count = {}'.format(len(json_data[:5000])))
# log_.info(f"json_data = {json_data}")
result = request_post(request_url=config_.NOTIFY_BACKEND_updateFallBackVideoList_URL,
request_data={'videos': json_data[:5000]})
if result is None:
log_.error('notify backend updateFallBackVideoList fail!')
elif result['code'] == 0:
log_.info('notify backend updateFallBackVideoList success!')
else:
log_.error('notify backend updateFallBackVideoList fail!')
# 去重更新rov模型结果，并另存为redis中
# initial_data_dup = {}
# for video_id, score in initial_data:
# if int(video_id) not in day_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_24H}{rule_key}.{now_dt}.{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 merge_df(df_left, df_right):
"""
df按照videoid 合并，对应特征求和
:param df_left:
:param df_right:
:return:
"""
df_merged = pd.merge(df_left, df_right, on=['videoid'], how='outer', suffixes=['_x', '_y'])
df_merged.fillna(0, inplace=True)
feature_list = ['videoid']
for feature in features:
if feature in ['apptype', 'videoid']:
continue
df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
feature_list.append(feature)
return df_merged[feature_list]
def merge_df_with_score(df_left, df_right):
"""
df 按照videoid合并，平台回流人数、回流人数、分数 分别求和
:param df_left:
:param df_right:
:return:
"""
df_merged = pd.merge(df_left, df_right, on=['videoid'], how='outer', suffixes=['_x', '_y'])
df_merged.fillna(0, inplace=True)
feature_list = ['videoid', '回流人数', 'platform_return', 'score']
for feature in feature_list[1:]:
df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
return df_merged[feature_list]
def rank_by_h(now_date, now_h, rule_params, project, table):
# 获取特征数据
feature_df = get_feature_data(now_date=now_date, now_h=now_h, project=project, table=table)
feature_df['apptype'] = feature_df['apptype'].astype(int)
# rank
data_params_item = rule_params.get('data_params')
rule_params_item = rule_params.get('rule_params')
"""
for param in rule_params.get('params_list'):
data_key = param.get('data')
data_param = data_params_item.get(data_key)
log_.info(f"data_key = {data_key}, data_param = {data_param}")
df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
df_merged = reduce(merge_df, df_list)
rule_key = param.get('rule')
rule_param = rule_params_item.get(rule_key)
log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
# 计算score
cal_score_func = rule_param.get('cal_score_func', 1)
if cal_score_func == 2:
score_df = cal_score2(df=df_merged, param=rule_param)
else:
score_df = cal_score1(df=df_merged)
video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
rule_key=rule_key, param=rule_param, data_key=data_key)
"""
for param in rule_params.get('params_list'):
score_df_list = []
notify_backend = param.get('notify_backend', False)
data_key = param.get('data')
data_param = data_params_item.get(data_key)
log_.info(f"data_key = {data_key}, data_param = {data_param}")
rule_key = param.get('rule')
rule_param = rule_params_item.get(rule_key)
log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
# cal_score_func = rule_param.get('cal_score_func', 1)
merge_func = rule_param.get('merge_func', 1)
if merge_func == 2:
for apptype, weight in data_param.items():
df = feature_df[feature_df['apptype'] == apptype]
# 计算score
score_df = cal_score(df=df, param=rule_param)
score_df['score'] = score_df['score'] * weight
score_df_list.append(score_df)
# 分数合并
df_merged = reduce(merge_df_with_score, score_df_list)
# 更新平台回流比
df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['回流人数']
video_rank_h(df=df_merged, now_date=now_date, now_h=now_h,
rule_key=rule_key, param=rule_param, data_key=data_key,
notify_backend=notify_backend)
else:
df_list = [feature_df[feature_df['apptype'] == apptype] for apptype, _ in data_param.items()]
df_merged = reduce(merge_df, df_list)
score_df = cal_score(df=df_merged, param=rule_param)
video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
rule_key=rule_key, param=rule_param, data_key=data_key,
notify_backend=notify_backend)
# # to-csv
# score_filename = f"score_by24h_{key}_{datetime.strftime(now_date, '%Y%m%d%H')}.csv"
# score_df.to_csv(f'./data/{score_filename}')
# # to-logs
# log_.info({"date": datetime.strftime(now_date, '%Y%m%d%H'),
# "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_BY_24H,
# "rule_key": key,
# # "score_df": score_df[['videoid', 'score']]
# })
def h_rank_bottom(now_date, now_h, rule_params):
"""未按时更新数据，用模型召回数据作为当前的数据"""
redis_helper = RedisHelper()
if now_h == 0:
redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
redis_h = 23
else:
redis_dt = datetime.strftime(now_date, '%Y%m%d')
redis_h = now_h - 1
key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_24H, config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER]
for param in rule_params.get('params_list'):
data_key = param.get('data')
rule_key = param.get('rule')
log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
for key_prefix in key_prefix_list:
key_name = f"{key_prefix}{data_key}:{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()
for video_id, score in initial_data:
final_data[video_id] = score
# 存入对应的redis
final_key_name = \
f"{key_prefix}{data_key}:{rule_key}:{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=2 * 3600)
"""
for app_type, params in rule_params.items():
log_.info(f"app_type = {app_type}")
for param in params.get('params_list'):
data_key = param.get('data')
rule_key = param.get('rule')
log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
for key_prefix in key_prefix_list:
key_name = f"{key_prefix}{app_type}:{data_key}:{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()
for video_id, score in initial_data:
final_data[video_id] = score
# 存入对应的redis
final_key_name = \
f"{key_prefix}{app_type}:{data_key}:{rule_key}:{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=2 * 3600)
# 清空线上过滤应用列表
# redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")
"""
def h_timer_check():
try:
project = config_.PROJECT_24H_APP_TYPE
table = config_.TABLE_24H_APP_TYPE
rule_params = config_.RULE_PARAMS_24H_APP_TYPE
now_date = datetime.today()
log_.info(f"now_date: {datetime.strftime(now_date, '%Y%m%d%H')}")
now_min = datetime.now().minute
now_h = datetime.now().hour
# 查看当前天级更新的数据是否已准备好
h_data_count = h_data_check(project=project, table=table, now_date=now_date, now_h=now_h)
if now_h == 23 or now_h < 8:
log_.info(f'now_h = {now_h} use bottom data!')
h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
log_.info(f"24h_data end!")
elif h_data_count > 0:
log_.info(f'h_by24h_data_count = {h_data_count}')
# 数据准备好，进行更新
rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params, project=project, table=table)
log_.info(f"24h_data end!")
elif now_min > 40:
log_.info('h_by24h_recall data is None, use bottom data!')
h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
log_.info(f"24h_data end!")
else:
# 数据没准备好，1分钟后重新检查
Timer(60, h_timer_check).start()
except Exception as e:
log_.error(f"不区分地域24h数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
send_msg_to_feishu(
webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
msg_text=f"rov-offline{config_.ENV_TEXT} - 不区分地域24h数据更新失败\n"
f"exception: {e}\n"
f"traceback: {traceback.format_exc()}"
)
if __name__ == '__main__':
log_.info(f"24h_data start...")
h_timer_check()