|
|
@@ -0,0 +1,708 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+# @ModuleName: region_rule_rank_h
|
|
|
+# @Author: Liqian
|
|
|
+# @Time: 2022/5/5 15:54
|
|
|
+# @Software: PyCharm
|
|
|
+import json
|
|
|
+import multiprocessing
|
|
|
+import os
|
|
|
+import sys
|
|
|
+import traceback
|
|
|
+
|
|
|
+import gevent
|
|
|
+import datetime
|
|
|
+import pandas as pd
|
|
|
+import math
|
|
|
+from functools import reduce
|
|
|
+from odps import ODPS
|
|
|
+from threading import Timer, Thread
|
|
|
+from utils import MysqlHelper, RedisHelper, get_data_from_odps, filter_video_status, filter_shield_video, \
|
|
|
+ check_table_partition_exits, filter_video_status_app, send_msg_to_feishu, filter_political_videos
|
|
|
+from config import set_config
|
|
|
+from log import Log
|
|
|
+from check_video_limit_distribute_new import update_limit_video_score
|
|
|
+
|
|
|
+# os.environ['NUMEXPR_MAX_THREADS'] = '16'
|
|
|
+
|
|
|
+config_, _ = set_config()
|
|
|
+log_ = Log()
|
|
|
+
|
|
|
+region_code = config_.REGION_CODE
|
|
|
+
|
|
|
+features = [
|
|
|
+ 'apptype',
|
|
|
+ 'code',
|
|
|
+ 'videoid',
|
|
|
+ 'lastonehour_preview', # 过去1小时预曝光人数
|
|
|
+ 'lastonehour_view', # 过去1小时曝光人数
|
|
|
+ 'lastonehour_play', # 过去1小时播放人数
|
|
|
+ 'lastonehour_share', # 过去1小时分享人数
|
|
|
+ 'lastonehour_return', # 过去1小时分享，过去1小时回流人数
|
|
|
+ 'lastonehour_preview_total', # 过去1小时预曝光次数
|
|
|
+ 'lastonehour_view_total', # 过去1小时曝光次数
|
|
|
+ 'lastonehour_play_total', # 过去1小时播放次数
|
|
|
+ 'lastonehour_share_total', # 过去1小时分享次数
|
|
|
+ 'platform_return',
|
|
|
+ 'lastonehour_show', # 不区分地域
|
|
|
+ 'lastonehour_show_region', # 地域分组
|
|
|
+]
|
|
|
+
|
|
|
+
|
|
|
+def data2file(data, filepath):
|
|
|
+ """数据写入文件"""
|
|
|
+ filedir = '/'.join(filepath.split('/')[:-1])
|
|
|
+ if not os.path.exists(filedir):
|
|
|
+ os.makedirs(filedir)
|
|
|
+ with open(filepath, 'w') as wf:
|
|
|
+ wf.write(data)
|
|
|
+
|
|
|
+
|
|
|
+def get_region_code(region):
|
|
|
+ """获取省份对应的code"""
|
|
|
+ mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO)
|
|
|
+ sql = f"SELECT ad_code FROM region_adcode WHERE parent_id = 0 AND region LIKE '{region}%';"
|
|
|
+ ad_code = mysql_helper.get_data(sql=sql)
|
|
|
+ return ad_code[0][0]
|
|
|
+
|
|
|
+
|
|
|
+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')
|
|
|
+ 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_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_day_30day_videos(now_date, data_key, rule_key):
|
|
|
+ """获取天级更新相对30天的视频id"""
|
|
|
+ redis_helper = RedisHelper()
|
|
|
+ day_30day_recall_key_prefix = config_.RECALL_KEY_NAME_PREFIX_30DAY
|
|
|
+ now_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
|
|
|
+ day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{now_dt}"
|
|
|
+ if not redis_helper.key_exists(key_name=day_30day_recall_key_name):
|
|
|
+ redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d')
|
|
|
+ day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{redis_dt}"
|
|
|
+ data = redis_helper.get_all_data_from_zset(key_name=day_30day_recall_key_name, with_scores=True)
|
|
|
+ if data is None:
|
|
|
+ return None
|
|
|
+ video_ids = [int(video_id) for video_id, _ in data]
|
|
|
+ return video_ids
|
|
|
+
|
|
|
+
|
|
|
+def get_feature_data(project, table, 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_preview+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) == 'video-show':
|
|
|
+ df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
|
|
|
+ elif param.get('view_type', None) == 'video-show-region':
|
|
|
+ df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
|
|
|
+ else:
|
|
|
+ df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
|
|
|
+ df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
|
|
|
+ df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
|
|
|
+
|
|
|
+ df['score1'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
|
|
|
+
|
|
|
+ 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 add_func1(initial_df, pre_h_df):
|
|
|
+ """当前小时级数据与前几个小时数据合并"""
|
|
|
+ score_list = initial_df['score'].to_list()
|
|
|
+ if len(score_list) > 0:
|
|
|
+ min_score = min(score_list)
|
|
|
+ else:
|
|
|
+ min_score = 0
|
|
|
+ pre_h_df = pre_h_df[pre_h_df['score'] > min_score]
|
|
|
+ df = pd.concat([initial_df, pre_h_df], ignore_index=True)
|
|
|
+ # videoid去重，保留分值高
|
|
|
+ df['videoid'] = df['videoid'].astype(int)
|
|
|
+ df = df.sort_values(by=['score'], ascending=False)
|
|
|
+ df = df.drop_duplicates(subset=['videoid'], keep="first")
|
|
|
+ return df
|
|
|
+
|
|
|
+
|
|
|
+def add_func2(initial_df, pre_h_df):
|
|
|
+ """当前小时级数据与前几个小时数据合并: 当前小时存在的视频以当前小时为准，否则以高分为主"""
|
|
|
+ score_list = initial_df['score'].to_list()
|
|
|
+ if len(score_list) > 0:
|
|
|
+ min_score = min(score_list)
|
|
|
+ else:
|
|
|
+ min_score = 0
|
|
|
+ initial_video_id_list = initial_df['videoid'].to_list()
|
|
|
+ pre_h_df = pre_h_df[pre_h_df['score'] > min_score]
|
|
|
+ pre_h_df = pre_h_df[~pre_h_df['videoid'].isin(initial_video_id_list)]
|
|
|
+
|
|
|
+ df = pd.concat([initial_df, pre_h_df], ignore_index=True)
|
|
|
+ # videoid去重，保留分值高
|
|
|
+ df['videoid'] = df['videoid'].astype(int)
|
|
|
+ df = df.sort_values(by=['score'], ascending=False)
|
|
|
+ df = df.drop_duplicates(subset=['videoid'], keep="first")
|
|
|
+ return df
|
|
|
+
|
|
|
+
|
|
|
+def add_videos(initial_df, now_date, rule_key, region, data_key, hour_count, top, add_func):
|
|
|
+ """
|
|
|
+ 地域小时级数据列表中增加前6h优质视频
|
|
|
+ :param initial_df: 地域小时级筛选结果
|
|
|
+ :param now_date:
|
|
|
+ :param data_key:
|
|
|
+ :param region:
|
|
|
+ :param rule_key:
|
|
|
+ :param hour_count: 前几个小时, type-int
|
|
|
+ :param top: type-int
|
|
|
+ :return: df
|
|
|
+ """
|
|
|
+ redis_helper = RedisHelper()
|
|
|
+ pre_h_data = []
|
|
|
+ for i in range(1, hour_count+1):
|
|
|
+ pre_date = now_date - datetime.timedelta(hours=i)
|
|
|
+ pre_h = pre_date.hour
|
|
|
+ pre_h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \
|
|
|
+ f"{datetime.datetime.strftime(pre_date, '%Y%m%d')}:{pre_h}"
|
|
|
+ pre_h_top_data = redis_helper.get_data_zset_with_index(key_name=pre_h_recall_key_name,
|
|
|
+ start=0, end=top-1,
|
|
|
+ desc=True, with_scores=True)
|
|
|
+ if pre_h_top_data is None:
|
|
|
+ continue
|
|
|
+ pre_h_data.extend(pre_h_top_data)
|
|
|
+ pre_h_df = pd.DataFrame(data=pre_h_data, columns=['videoid', 'score'])
|
|
|
+ if add_func == 'func2':
|
|
|
+ df = add_func2(initial_df=initial_df, pre_h_df=pre_h_df)
|
|
|
+ else:
|
|
|
+ df = add_func1(initial_df=initial_df, pre_h_df=pre_h_df)
|
|
|
+ return df
|
|
|
+
|
|
|
+
|
|
|
+def video_rank(df, now_date, now_h, rule_key, param, region, data_key, add_videos_with_pre_h=False, hour_count=0):
|
|
|
+ """
|
|
|
+ 获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
|
|
|
+ :param hour_count:
|
|
|
+ :param add_videos_with_pre_h:
|
|
|
+ :param data_key:
|
|
|
+ :param df:
|
|
|
+ :param now_date:
|
|
|
+ :param now_h:
|
|
|
+ :param rule_key: 小时级数据进入条件
|
|
|
+ :param param: 小时级数据进入条件参数
|
|
|
+ :param region: 所属地域
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ redis_helper = RedisHelper()
|
|
|
+
|
|
|
+ # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
|
|
|
+ return_count = param.get('return_count', 1)
|
|
|
+ score_value = param.get('score_rule', 0)
|
|
|
+ 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)]
|
|
|
+
|
|
|
+ # videoid重复时，保留分值高
|
|
|
+ h_recall_df = h_recall_df.sort_values(by=['score'], ascending=False)
|
|
|
+ h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first')
|
|
|
+ h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
|
|
|
+
|
|
|
+ # 增加打捞的优质视频
|
|
|
+ if add_videos_with_pre_h is True:
|
|
|
+ add_func = param.get('add_func', None)
|
|
|
+ h_recall_df = add_videos(initial_df=h_recall_df, now_date=now_date, rule_key=rule_key,
|
|
|
+ region=region, data_key=data_key, hour_count=hour_count, top=10, add_func=add_func)
|
|
|
+
|
|
|
+ h_recall_videos = h_recall_df['videoid'].to_list()
|
|
|
+ # log_.info(f'h_recall videos count = {len(h_recall_videos)}')
|
|
|
+
|
|
|
+ # 视频状态过滤
|
|
|
+ if data_key in ['data7', ]:
|
|
|
+ filtered_videos = filter_video_status_app(h_recall_videos)
|
|
|
+ else:
|
|
|
+ filtered_videos = filter_video_status(h_recall_videos)
|
|
|
+ # log_.info('filtered_videos count = {}'.format(len(filtered_videos)))
|
|
|
+
|
|
|
+ # 屏蔽视频过滤
|
|
|
+ shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
|
|
|
+ shield_key_name_list = shield_config.get(region, None)
|
|
|
+ if shield_key_name_list is not None:
|
|
|
+ filtered_videos = filter_shield_video(video_ids=filtered_videos, shield_key_name_list=shield_key_name_list)
|
|
|
+ # log_.info(f"shield filtered_videos count = {len(filtered_videos)}")
|
|
|
+
|
|
|
+ # 涉政视频过滤
|
|
|
+ political_filter = param.get('political_filter', None)
|
|
|
+ if political_filter is True:
|
|
|
+ log_.info(f"political filter videos count = {len(filtered_videos)}")
|
|
|
+ filtered_videos = filter_political_videos(video_ids=filtered_videos)
|
|
|
+ log_.info(f"political filtered videos count = {len(filtered_videos)}")
|
|
|
+
|
|
|
+ h_video_ids = []
|
|
|
+ h_recall_result_mapping = {}
|
|
|
+ for video_id in filtered_videos:
|
|
|
+ score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
|
|
|
+ # print(score)
|
|
|
+ h_recall_result_mapping[int(video_id)] = float(score)
|
|
|
+ h_video_ids.append(int(video_id))
|
|
|
+ # 限流视频score调整
|
|
|
+ h_recall_result_mapping = update_limit_video_score(initial_videos=h_recall_result_mapping)
|
|
|
+ if h_recall_result_mapping:
|
|
|
+ # 按照score排序
|
|
|
+ h_recall_result = [(int(vid), float(score)) for vid, score in h_recall_result_mapping.items()]
|
|
|
+ h_recall_result = sorted(h_recall_result, key=lambda x: x[1], reverse=True)
|
|
|
+ h_recall_result = [(vid, round(score, 10)) for vid, score in h_recall_result]
|
|
|
+ h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}"
|
|
|
+ log_.info(f"h_recall_result count = {len(h_recall_result)}")
|
|
|
+ # 写入对应的redis
|
|
|
+ redis_helper.set_data_to_redis(
|
|
|
+ key_name=h_recall_key_name, value=json.dumps(h_recall_result[:10000]), expire_time=30 * 24 * 3600
|
|
|
+ )
|
|
|
+ # 写入本地文件
|
|
|
+ filename = f"{region}_{data_key}_{rule_key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.txt"
|
|
|
+ data2file(data=json.dumps(h_recall_result), filepath=f"./data/region_h/{filename}")
|
|
|
+
|
|
|
+ region_24h_rule_key = param.get('region_24h_rule_key', 'rule1')
|
|
|
+ by_24h_rule_key = param.get('24h_rule_key', None)
|
|
|
+ # 与其他召回视频池去重，存入对应的redis
|
|
|
+ dup_to_redis(now_date=now_date, now_h=now_h, rule_key=rule_key,
|
|
|
+ region_24h_rule_key=region_24h_rule_key, by_24h_rule_key=by_24h_rule_key,
|
|
|
+ region=region, data_key=data_key, political_filter=political_filter,
|
|
|
+ shield_config=shield_config)
|
|
|
+
|
|
|
+
|
|
|
+def dup_data(initial_key_name, dup_key_name, region, political_filter, shield_config, filepath):
|
|
|
+ redis_helper = RedisHelper()
|
|
|
+ if redis_helper.key_exists(key_name=initial_key_name):
|
|
|
+ initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True)
|
|
|
+ # 屏蔽视频过滤
|
|
|
+ initial_video_ids = [int(video_id) for video_id, _ in initial_data]
|
|
|
+ shield_key_name_list = shield_config.get(region, None)
|
|
|
+ if shield_key_name_list is not None:
|
|
|
+ initial_video_ids = filter_shield_video(video_ids=initial_video_ids,
|
|
|
+ shield_key_name_list=shield_key_name_list)
|
|
|
+ # 涉政视频过滤
|
|
|
+ if political_filter is True:
|
|
|
+ initial_video_ids = filter_political_videos(video_ids=initial_video_ids)
|
|
|
+
|
|
|
+ dup_data = {}
|
|
|
+ for video_id, score in initial_data:
|
|
|
+ if int(video_id) in initial_video_ids:
|
|
|
+ dup_data[int(video_id)] = score
|
|
|
+ # 限流视频score调整
|
|
|
+ dup_data = update_limit_video_score(initial_videos=dup_data)
|
|
|
+ if dup_data:
|
|
|
+ # 按照score排序
|
|
|
+ data = [(int(vid), float(score)) for vid, score in dup_data.items()]
|
|
|
+ data = sorted(data, key=lambda x: x[1], reverse=True)
|
|
|
+ data = [(vid, round(score, 10)) for vid, score in data]
|
|
|
+
|
|
|
+ log_.info(f"data count = {len(data)}")
|
|
|
+ # 写入对应的redis
|
|
|
+ redis_helper.set_data_to_redis(
|
|
|
+ key_name=dup_key_name, value=json.dumps(data[:10000]), expire_time=30 * 24 * 3600
|
|
|
+ )
|
|
|
+ # 写入本地文件
|
|
|
+ data2file(data=json.dumps(data), filepath=filepath)
|
|
|
+
|
|
|
+
|
|
|
+def dup_to_redis(now_date, now_h, rule_key, region_24h_rule_key, by_24h_rule_key,
|
|
|
+ region, data_key, political_filter, shield_config):
|
|
|
+ """将地域分组小时级数据与其他召回视频池去重，存入对应的redis"""
|
|
|
+ # ##### 更新地域分组小时级24h列表，并另存为redis中
|
|
|
+ region_24h_key_name = \
|
|
|
+ f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}:{data_key}:{region_24h_rule_key}:" \
|
|
|
+ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
|
|
|
+ region_24h_dup_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{data_key}:{rule_key}"
|
|
|
+ filename = f"{region}_{data_key}_{rule_key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.txt"
|
|
|
+ dup_data(initial_key_name=region_24h_key_name, dup_key_name=region_24h_dup_key_name, region=region,
|
|
|
+ political_filter=political_filter, shield_config=shield_config, filepath=f"./data/region24h/{filename}")
|
|
|
+
|
|
|
+ # ##### 小程序相对24h更新结果，并另存为redis中
|
|
|
+ h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{by_24h_rule_key}:" \
|
|
|
+ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
|
|
|
+ h_24h_dup_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{data_key}:{rule_key}"
|
|
|
+ filename = f"{region}_{data_key}_{rule_key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.txt"
|
|
|
+ dup_data(initial_key_name=h_24h_key_name, dup_key_name=h_24h_dup_key_name, region=region,
|
|
|
+ political_filter=political_filter, shield_config=shield_config, filepath=f"./data/24h/{filename}")
|
|
|
+
|
|
|
+ # ##### 去重小程序相对24h 筛选后剩余数据 更新结果，并另存为redis中
|
|
|
+ other_h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:" \
|
|
|
+ f"{by_24h_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
|
|
|
+ other_h_24h_dup_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{data_key}:{rule_key}"
|
|
|
+ filename = f"{region}_{data_key}_{rule_key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.txt"
|
|
|
+ dup_data(initial_key_name=other_h_24h_key_name, dup_key_name=other_h_24h_dup_key_name, region=region,
|
|
|
+ political_filter=political_filter, shield_config=shield_config, filepath=f"./data/24h_other/{filename}")
|
|
|
+
|
|
|
+
|
|
|
+def merge_df(df_left, df_right):
|
|
|
+ """
|
|
|
+ df按照videoid, code 合并，对应特征求和
|
|
|
+ :param df_left:
|
|
|
+ :param df_right:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
|
|
|
+ df_merged.fillna(0, inplace=True)
|
|
|
+ feature_list = ['videoid', 'code']
|
|
|
+ for feature in features:
|
|
|
+ if feature in ['apptype', 'videoid', 'code']:
|
|
|
+ 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, code]合并，平台回流人数、回流人数、分数 分别求和
|
|
|
+ :param df_left:
|
|
|
+ :param df_right:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
|
|
|
+ df_merged.fillna(0, inplace=True)
|
|
|
+ feature_list = ['videoid', 'code', 'lastonehour_return', 'platform_return', 'score']
|
|
|
+ for feature in feature_list[2:]:
|
|
|
+ df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
|
|
|
+ return df_merged[feature_list]
|
|
|
+
|
|
|
+
|
|
|
+def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
|
|
|
+ add_videos_with_pre_h, hour_count):
|
|
|
+ log_.info(f"region = {region} start...")
|
|
|
+ # 计算score
|
|
|
+ region_df = df_merged[df_merged['code'] == region]
|
|
|
+ log_.info(f'region = {region}, region_df count = {len(region_df)}')
|
|
|
+ score_df = cal_score(df=region_df, param=rule_param)
|
|
|
+ video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=rule_key, param=rule_param,
|
|
|
+ region=region, data_key=data_key,
|
|
|
+ add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count)
|
|
|
+ log_.info(f"region = {region} end!")
|
|
|
+
|
|
|
+
|
|
|
+def process_with_region2(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
|
|
|
+ add_videos_with_pre_h, hour_count):
|
|
|
+ log_.info(f"region = {region} start...")
|
|
|
+ region_score_df = df_merged[df_merged['code'] == region]
|
|
|
+ log_.info(f'region = {region}, region_score_df count = {len(region_score_df)}')
|
|
|
+ video_rank(df=region_score_df, now_date=now_date, now_h=now_h, region=region,
|
|
|
+ rule_key=rule_key, param=rule_param, data_key=data_key,
|
|
|
+ add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count)
|
|
|
+ log_.info(f"region = {region} end!")
|
|
|
+
|
|
|
+
|
|
|
+def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag):
|
|
|
+ log_.info(f"app_type = {app_type} start...")
|
|
|
+ data_params_item = params.get('data_params')
|
|
|
+ rule_params_item = params.get('rule_params')
|
|
|
+ task_list = []
|
|
|
+ for param in 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}")
|
|
|
+ task_list.extend(
|
|
|
+ [
|
|
|
+ gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
|
|
|
+ now_date, now_h, rule_rank_h_flag)
|
|
|
+ for region in region_code_list
|
|
|
+ ]
|
|
|
+ )
|
|
|
+ gevent.joinall(task_list)
|
|
|
+ log_.info(f"app_type = {app_type} end!")
|
|
|
+
|
|
|
+
|
|
|
+ # log_.info(f"app_type = {app_type}")
|
|
|
+ # task_list = []
|
|
|
+ # for data_key, data_param in params['data_params'].items():
|
|
|
+ # 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)
|
|
|
+ # for rule_key, rule_param in params['rule_params'].items():
|
|
|
+ # log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
|
|
|
+ # task_list.extend(
|
|
|
+ # [
|
|
|
+ # gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
|
|
|
+ # now_date, now_h)
|
|
|
+ # for region in region_code_list
|
|
|
+ # ]
|
|
|
+ # )
|
|
|
+ # gevent.joinall(task_list)
|
|
|
+
|
|
|
+
|
|
|
+def copy_data_for_city(region, city_code, data_key, rule_key, shield_config, now_date):
|
|
|
+ """copy 对应数据到城市对应redis，并做相应屏蔽视频过滤"""
|
|
|
+ log_.info(f"city_code = {city_code} start ...")
|
|
|
+ redis_helper = RedisHelper()
|
|
|
+ key_prefix_list = [
|
|
|
+ (config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H, 'region_h'), # 地域小时级
|
|
|
+ (config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H, 'region_24h'), # 地域相对24h
|
|
|
+ (config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H, '24h'), # 不区分地域相对24h
|
|
|
+ (config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H, '24h_other') # 不区分地域相对24h筛选后
|
|
|
+ ]
|
|
|
+ for key_prefix, file_folder in key_prefix_list:
|
|
|
+ region_key = f"{key_prefix}{region}:{data_key}:{rule_key}"
|
|
|
+ city_key = f"{key_prefix}{city_code}:{data_key}:{rule_key}"
|
|
|
+ if not redis_helper.key_exists(key_name=region_key):
|
|
|
+ continue
|
|
|
+ region_data = redis_helper.get_data_from_redis(key_name=region_key)
|
|
|
+ if not region_data:
|
|
|
+ continue
|
|
|
+ # 屏蔽视频过滤
|
|
|
+ region_video_ids = [int(video_id) for video_id, _ in json.loads(region_data)]
|
|
|
+ shield_key_name_list = shield_config.get(city_code, None)
|
|
|
+ # shield_key_name_list = config_.SHIELD_CONFIG.get(city_code, None)
|
|
|
+ if shield_key_name_list is not None:
|
|
|
+ filtered_video_ids = filter_shield_video(video_ids=region_video_ids,
|
|
|
+ shield_key_name_list=shield_key_name_list)
|
|
|
+ else:
|
|
|
+ filtered_video_ids = region_video_ids
|
|
|
+ city_data = []
|
|
|
+ for video_id, score in json.loads(region_data):
|
|
|
+ if int(video_id) in filtered_video_ids:
|
|
|
+ city_data.append((int(video_id), score))
|
|
|
+
|
|
|
+ if len(city_data) > 0:
|
|
|
+ redis_helper.set_data_to_redis(key_name=city_key, value=json.dumps(city_data), expire_time=30 * 24 * 3600)
|
|
|
+ # 写入本地文件
|
|
|
+ filename = f"{region}_{data_key}_{rule_key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.txt"
|
|
|
+ data2file(data=json.dumps(city_data), filepath=f"./data/{file_folder}/{filename}")
|
|
|
+
|
|
|
+ log_.info(f"city_code = {city_code} end!")
|
|
|
+
|
|
|
+
|
|
|
+def process_with_param(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h):
|
|
|
+ log_.info(f"param = {param} start...")
|
|
|
+
|
|
|
+ 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}")
|
|
|
+ merge_func = rule_param.get('merge_func', None)
|
|
|
+ # 是否在地域小时级数据中增加打捞的优质视频
|
|
|
+ add_videos_with_pre_h = rule_param.get('add_videos_with_pre_h', False)
|
|
|
+ hour_count = rule_param.get('hour_count', 0)
|
|
|
+
|
|
|
+ if merge_func == 2:
|
|
|
+ score_df_list = []
|
|
|
+ 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['lastonehour_return']
|
|
|
+ task_list = [
|
|
|
+ gevent.spawn(process_with_region2,
|
|
|
+ region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
|
|
|
+ add_videos_with_pre_h, hour_count)
|
|
|
+ for region in region_code_list
|
|
|
+ ]
|
|
|
+ else:
|
|
|
+ df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
|
|
|
+ df_merged = reduce(merge_df, df_list)
|
|
|
+ task_list = [
|
|
|
+ gevent.spawn(process_with_region,
|
|
|
+ region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
|
|
|
+ add_videos_with_pre_h, hour_count)
|
|
|
+ for region in region_code_list
|
|
|
+ ]
|
|
|
+
|
|
|
+ gevent.joinall(task_list)
|
|
|
+
|
|
|
+ # 特殊城市视频数据准备
|
|
|
+ # 屏蔽视频过滤
|
|
|
+ shield_config = rule_param.get('shield_config', config_.SHIELD_CONFIG)
|
|
|
+ for region, city_list in config_.REGION_CITY_MAPPING.items():
|
|
|
+ t = [
|
|
|
+ gevent.spawn(
|
|
|
+ copy_data_for_city,
|
|
|
+ region, city_code, data_key, rule_key, shield_config, now_date
|
|
|
+ )
|
|
|
+ for city_code in city_list
|
|
|
+ ]
|
|
|
+ gevent.joinall(t)
|
|
|
+
|
|
|
+ log_.info(f"param = {param} end!")
|
|
|
+
|
|
|
+
|
|
|
+def rank_by_h(project, table, now_date, now_h, rule_params, region_code_list):
|
|
|
+ # 获取特征数据
|
|
|
+ feature_df = get_feature_data(project=project, table=table, now_date=now_date)
|
|
|
+ feature_df['apptype'] = feature_df['apptype'].astype(int)
|
|
|
+ data_params_item = rule_params.get('data_params')
|
|
|
+ rule_params_item = rule_params.get('rule_params')
|
|
|
+ params_list = rule_params.get('params_list_new')
|
|
|
+ pool = multiprocessing.Pool(processes=len(params_list))
|
|
|
+ for param in params_list:
|
|
|
+ pool.apply_async(
|
|
|
+ func=process_with_param,
|
|
|
+ args=(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h)
|
|
|
+ )
|
|
|
+ pool.close()
|
|
|
+ pool.join()
|
|
|
+
|
|
|
+
|
|
|
+def h_bottom_process(param, rule_params_item, region_code_list, now_date, now_h):
|
|
|
+ data_key = param.get('data')
|
|
|
+ rule_key = param.get('rule')
|
|
|
+ rule_param = rule_params_item.get(rule_key)
|
|
|
+ log_.info(f"data_key = {data_key}, rule_key = {rule_key}, rule_param = {rule_param}")
|
|
|
+ region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1')
|
|
|
+ by_24h_rule_key = rule_param.get('24h_rule_key', None)
|
|
|
+ # 涉政视频过滤
|
|
|
+ political_filter = param.get('political_filter', None)
|
|
|
+ # 屏蔽视频过滤
|
|
|
+ shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
|
|
|
+ for region in region_code_list:
|
|
|
+ log_.info(f"region = {region}")
|
|
|
+ # 与其他召回视频池去重，存入对应的redis
|
|
|
+ dup_to_redis(now_date=now_date, now_h=now_h, rule_key=rule_key,
|
|
|
+ region_24h_rule_key=region_24h_rule_key, region=region,
|
|
|
+ data_key=data_key, by_24h_rule_key=by_24h_rule_key,
|
|
|
+ political_filter=political_filter, shield_config=shield_config)
|
|
|
+ # 特殊城市视频数据准备
|
|
|
+ for region, city_list in config_.REGION_CITY_MAPPING.items():
|
|
|
+ t = [
|
|
|
+ gevent.spawn(
|
|
|
+ copy_data_for_city,
|
|
|
+ region, city_code, data_key, rule_key, shield_config, now_date
|
|
|
+ )
|
|
|
+ for city_code in city_list
|
|
|
+ ]
|
|
|
+ gevent.joinall(t)
|
|
|
+
|
|
|
+
|
|
|
+def h_rank_bottom(now_date, now_h, rule_params, region_code_list):
|
|
|
+ """未按时更新数据，用上一小时结果作为当前小时的数据"""
|
|
|
+ # 以上一小时的地域分组数据作为当前小时的数据
|
|
|
+ rule_params_item = rule_params.get('rule_params')
|
|
|
+ params_list = rule_params.get('params_list_new')
|
|
|
+ pool = multiprocessing.Pool(processes=len(params_list))
|
|
|
+ for param in params_list:
|
|
|
+ pool.apply_async(
|
|
|
+ func=h_bottom_process,
|
|
|
+ args=(param, rule_params_item, region_code_list, now_date, now_h)
|
|
|
+ )
|
|
|
+ pool.close()
|
|
|
+ pool.join()
|
|
|
+
|
|
|
+
|
|
|
+def h_timer_check():
|
|
|
+ try:
|
|
|
+ rule_params = config_.RULE_PARAMS_REGION_APP_TYPE
|
|
|
+ project = config_.PROJECT_REGION_APP_TYPE
|
|
|
+ table = config_.TABLE_REGION_APP_TYPE
|
|
|
+ region_code_list = [code for region, code in region_code.items()]
|
|
|
+ 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:
|
|
|
+ h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list)
|
|
|
+ log_.info(f"region_h_data end!")
|
|
|
+ return
|
|
|
+ # 查看当前小时更新的数据是否已准备好
|
|
|
+ h_data_count = h_data_check(project=project, table=table, now_date=now_date)
|
|
|
+ if h_data_count > 0:
|
|
|
+ log_.info(f'region_h_data_count = {h_data_count}')
|
|
|
+ # 数据准备好，进行更新
|
|
|
+ rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params,
|
|
|
+ project=project, table=table, region_code_list=region_code_list)
|
|
|
+ log_.info(f"region_h_data end!")
|
|
|
+ elif now_min > 40:
|
|
|
+ log_.info('h_recall data is None, use bottom data!')
|
|
|
+ h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list)
|
|
|
+ log_.info(f"region_h_data end!")
|
|
|
+ else:
|
|
|
+ # 数据没准备好，1分钟后重新检查
|
|
|
+ Timer(60, h_timer_check).start()
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ log_.error(f"地域分组小时级数据更新失败, 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} - 地域分组小时级数据更新失败\n"
|
|
|
+ f"exception: {e}\n"
|
|
|
+ f"traceback: {traceback.format_exc()}"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ log_.info(f"region_h_data start...")
|
|
|
+ h_timer_check()
|
