rov-offline/rule_rank_day.py

import pandas as pd
from odps import ODPS
from datetime import datetime, timedelta
from threading import Timer
from utils import get_data_from_odps
from db_helper import RedisHelper
from config import set_config
from log import Log

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

features = [
    'videoid',
    'preview人数',  # 过去1天预曝光人数
    'view人数',  # 过去1天曝光人数
    'play人数',  # 过去1天播放人数
    'share人数',  # 过去1天分享人数
    '回流人数',  # 过去1天分享，过去1天回流人数
    'preview次数',  # 过去1天预曝光次数
    'view次数',  # 过去1天曝光次数
    'play次数',  # 过去1天播放次数
    'share次数',  # 过去1天分享次数
]


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}'
    key_dt = 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}'
        key_dt = pre_dt
    return key_name, key_dt


def day_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.strftime((now_date - timedelta(days=1)), '%Y%m%d')
        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.strftime((now_date - timedelta(days=1)), '%Y%m%d')
    # 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):
    pass


def video_rank_day(df, now_date, rule_key, param):
    """
    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
    :param df:
    :param now_date:
    :param rule_key: 天级规则数据进入条件
    :param param: 天级规则数据进入条件参数
    :return:
    """
    # 获取rov模型结果
    redis_helper = RedisHelper()
    key_name, key_dt = get_rov_redis_key(now_date=now_date)
    initial_data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1, with_scores=True)
    log_.info(f'initial data count = {len(initial_data)}, key_dt = {key_dt}')

    # 获取符合进入召回源条件的视频
    return_count = param.get('return_count')
    h_recall_df = df[df['回流人数'] > return_count]
    h_recall_videos = h_recall_df['videoid'].to_list()
    log_.info(f'h_recall videos count = {len(h_recall_videos)}')
    # 写入对应的redis
    h_video_ids =[]
    h_recall_result = {}
    for video_id in h_recall_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_DAY}{rule_key}.{datetime.strftime(now_date, '%Y%m%d')}"
    if len(h_recall_result) > 0:
        redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=7 * 24 * 3600)

    # 去重更新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)}")

    now_dt = datetime.strftime(now_date, '%Y%m%d')
    if key_dt == now_dt:
        initial_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_NOW
    else:
        initial_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE
    initial_key_name = f"{initial_key_name_prefix}{rule_key}.{now_dt}"
    if len(initial_data_dup) > 0:
        redis_helper.add_data_with_zset(key_name=initial_key_name, data=initial_data_dup, expire_time=7 * 24 * 3600)


def rank_by_day(now_date, rule_params, project, table):
    # 获取特征数据
    feature_df = get_feature_data(now_date=now_date, project=project, table=table)
    # rank
    for key, value in rule_params.items():
        log_.info(f"rule = {key}, param = {value}")
        # 计算score
        cal_score_func = value.get('cal_score_func', 1)
        if cal_score_func == 2:
            score_df = cal_score2(df=feature_df)
        else:
            score_df = cal_score1(df=feature_df)
        video_rank_day(df=score_df, now_date=now_date, rule_key=key, param=value)
        # to-csv
        score_filename = f"score_{key}_{datetime.strftime(now_date, '%Y%m%d')}.csv"
        score_df.to_csv(f'./data/{score_filename}')


def day_rank_bottom(now_date, rule_key):
    """未按时更新数据，用模型召回数据作为当前的数据"""
    log_.info(f"rule_key = {rule_key}")
    now_dt = datetime.strftime(now_date, '%Y%m%d')
    redis_helper = RedisHelper()
    key_name, key_dt = get_rov_redis_key(now_date=now_date)
    initial_data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1, with_scores=True)
    final_data = dict()
    for video_id, score in initial_data:
        final_data[video_id] = score
    if key_dt == now_dt:
        key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_NOW
    else:
        key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE
    key_name = f"{key_name_prefix}{rule_key}.{now_dt}"
    if len(final_data) > 0:
        redis_helper.add_data_with_zset(key_name=key_name, data=final_data, expire_time=7 * 24 * 3600)


def day_timer_check():
    project = config_.PROJECT_DAY
    table = config_.TABLE_DAY
    rule_params = config_.RULE_PARAMS_DAY
    # return_count_list = [20, 10]
    now_date = datetime.today()
    log_.info(f"now_date: {datetime.strftime(now_date, '%Y%m%d')}")
    now_h = datetime.now().hour
    now_min = datetime.now().minute
    # 查看当前天级更新的数据是否已准备好
    h_data_count = day_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_day(now_date=now_date, rule_params=rule_params, project=project, table=table)
    elif now_min > 50:
        log_.info('day_recall data is None!')
        for key, _ in rule_params.items():
            day_rank_bottom(now_date=now_date, rule_key=key)
    else:
        # 数据没准备好，1分钟后重新检查
        Timer(60, day_timer_check).start()


if __name__ == '__main__':
    day_timer_check()