rov-offline/rule_rank_h_18_19.py

# -*- coding: utf-8 -*-
# @ModuleName: rule_rank_h_18_19
# @Author: Liqian
# @Time: 2022/4/21 下午4:31
# @Software: PyCharm

import time
import datetime
import pandas as pd
import math
import random
from odps import ODPS
from threading import Timer
from get_data import get_data_from_odps
from db_helper import RedisHelper, MysqlHelper
from my_config import set_config
from log import Log
from my_utils import filter_video_status

config_, env = set_config()
log_ = Log()

features = [
    'videoid',
    'lastonehour_view',  # 过去1小时曝光
    'lastonehour_play',  # 过去1小时播放
    'lastonehour_share',  # 过去1小时分享
    'lastonehour_return',  # 过去1小时分享，过去1小时回流
]


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_feature_data(now_date, project, table):
    """获取特征数据"""
    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):
    """
    计算score
    :param df: 特征数据
    :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

    # 视频状态过滤
    log_.info(f'initial_df count = {len(df)}')
    video_ids = [int(video_id) for video_id in df['videoid']]
    filtered_result = filter_video_status(video_ids=video_ids)
    filter_result = set(video_ids) - set(filtered_result)
    df['videoid'] = df['videoid'].astype(int)
    filter_df = df[df['videoid'].isin(filter_result)]
    df = df.append(filter_df)
    df = df.drop_duplicates(['videoid'], keep=False)
    log_.info(f'filtered_df count = {len(df)}')
    # 计算score
    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)
    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 = df.sort_values(by=['score'], ascending=False)
    return df


def video_rank(app_type, df, now_date, now_h, return_count):
    """
    根据回流数量，对视频进行二次排序
    :param app_type:
    :param df:
    :param now_date:
    :param now_h:
    :param return_count: 小时级数据回流限制数
    :return:
    """
    log_.info(f'df length = {len(df)}')
    # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
    h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= 0.005)]
    h_recall_videos = h_recall_df['videoid'].to_list()
    log_.info(f'h_recall videos count = {len(h_recall_videos)}')
    # 不符合进入召回源条件的视频
    df = df.append(h_recall_df)
    h_else_df = df.drop_duplicates(['videoid'], keep=False)
    h_else_df = h_else_df.sort_values(by=['score'], ascending=False)
    h_else_videos = h_else_df['videoid'].to_list()
    # 合并，给定分数
    final_videos = h_recall_videos + h_else_videos
    final_result = {}
    step = round(100/len(final_videos), 3)
    for i, video_id in enumerate(final_videos):
        score = 100 - i * step
        final_result[int(video_id)] = score
    # 写入对应的redis
    key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if len(final_result) > 0:
        redis_helper = RedisHelper()
        redis_helper.add_data_with_zset(key_name=key_name, data=final_result, expire_time=23 * 3600)


def rank_by_h(app_type, now_date, now_h, return_count_list, project, table):
    # 获取特征数据
    feature_df = get_feature_data(now_date=now_date, project=project, table=table)
    # 计算score
    score_df = cal_score(df=feature_df)
    # rank
    for cnt in return_count_list:
        log_.info(f"return_count = {cnt}")
        video_rank(app_type=app_type, df=score_df, now_date=now_date, now_h=now_h, return_count=cnt)
    # to-csv
    score_filename = f"score_{app_type}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
    score_df.to_csv(f'./data/{score_filename}')


def h_rank_bottom(app_type, now_date, now_h):
    """未按时更新数据，用上一小时结果作为当前小时的数据"""
    log_.info(f"app_type = {app_type}")
    # 获取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_name = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{redis_dt}.{redis_h}"
    initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
    final_data = dict()
    for video_id, score in initial_data:
        final_data[video_id] = score
    # 存入对应的redis
    final_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{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)


def h_timer_check(app_type):
    log_.info(f"app_type = {app_type}")
    return_count_list = [20]
    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(app_type=app_type, now_date=now_date, now_h=now_h)
        return
    # 查看当前小时更新的数据是否已准备好
    project = config_.PREDICT_PROJECT_18_19[str(app_type)]
    table = config_.PREDICT_TABLE_18_19[str(app_type)]
    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(app_type=app_type, now_date=now_date, now_h=now_h,
                  return_count_list=return_count_list, project=project, table=table)
    elif now_min > 50:
        log_.info('h_recall data is None, use bottom data!')
        h_rank_bottom(app_type=app_type, now_date=now_date, now_h=now_h)
    else:
        # 数据没准备好，1分钟后重新检查
        Timer(60, h_timer_check, args=[app_type]).start()


def predict(app_type_list):
    for app_type in app_type_list:
        h_timer_check(app_type=app_type)


def predict_test(app_type_list, count):
    now_date = datetime.datetime.today()
    now_h = datetime.datetime.now().hour
    log_.info(f"now_date = {datetime.datetime.strftime(now_date, '%Y%m%d%H')}, now_h = {now_h}")
    # 获取测试环境中最近发布的40000条视频
    sql = "SELECT id FROM wx_video ORDER BY id DESC LIMIT 40000;"
    mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO)
    data = mysql_helper.get_data(sql=sql)
    video_ids = [int(video[0]) for video in data]
    # 视频状态过滤
    filtered_videos = filter_video_status(video_ids)
    log_.info('filtered_videos count = {}'.format(len(filtered_videos)))

    for app_type in app_type_list:
        log_.info(f"app_type = {app_type}")
        videos_temp = random.sample(filtered_videos, count)
        redis_data_temp = {}
        csv_data_temp = []
        for video_id in videos_temp:
            score = random.uniform(0, 100)
            redis_data_temp[video_id] = score
            csv_data_temp.append({'video_id': video_id, 'rov_score': score})
        # 打包预测结果存入csv
        score_df = pd.DataFrame(data=csv_data_temp, columns=['video_id', 'rov_score'])
        score_df = score_df.sort_values(by=['rov_score'], ascending=False)
        score_filename = f"score_{app_type}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
        score_df.to_csv(f'./data/{score_filename}', index=False)

        # 存入对应的redis
        key_name = \
            f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        redis_helper = RedisHelper()
        redis_helper.add_data_with_zset(key_name=key_name, data=redis_data_temp)
        log_.info('data to redis finished!')


if __name__ == '__main__':
    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
    log_.info(f'appType: {app_type_list} predict start...')
    predict_start = time.time()
    if env in ['dev', 'test']:
        predict_test(app_type_list=app_type_list, count=300)
    elif env in ['pre', 'pro']:
        predict(app_type_list=app_type_list)
    else:
        log_.error('env error')
    predict_end = time.time()
    log_.info(f'appType: {app_type_list} predict end, execute time = {(predict_end - predict_start) * 1000}ms')