rov-offline/laohaokan_recommend_update.py

import datetime
import gevent
from db_helper import RedisHelper
from utils import send_msg_to_feishu
from config import set_config
from log import Log

config_, env = set_config()
log_ = Log()
initial_param = {'data': 'data1', 'rule': 'rule4'}
new_param = config_.LHK_RULE_PARAMS
redis_helper = RedisHelper()


def get_religion_videos(now_date):
    """获取宗教视频列表"""
    religion_key_name = f"{config_.KEY_NAME_PREFIX_RELIGION_VIDEOS}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
    if not redis_helper.key_exists(religion_key_name):
        redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d')
        religion_key_name = f"{config_.KEY_NAME_PREFIX_RELIGION_VIDEOS}{redis_dt}"
    religion_videos = redis_helper.get_all_data_from_zset(key_name=religion_key_name, desc=True, with_scores=True)
    if religion_videos is None:
        return []
    return religion_videos


def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_list):
    initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True)
    if initial_data is None or len(initial_data) == 0:
        return now_videos, religion_video_id_list

    initial_video_ids = [int(video_id) for video_id, _ in initial_data]
    initial_video_ids = [video_id for video_id in initial_video_ids if video_id not in now_videos]
    religion_video_id_list = [video_id for video_id in religion_video_id_list if video_id not in initial_video_ids]
    if len(religion_video_id_list) == 0:
        new_video_ids = initial_video_ids
    else:
        new_video_ids = []
        for i, video_id in enumerate(initial_video_ids):
            new_video_ids.append(video_id)
            now_videos.append(video_id)
            if i % 2 == 1 and len(religion_video_id_list) > 0:
                new_video_ids.append(religion_video_id_list[0])
                now_videos.append(religion_video_id_list[0])
                religion_video_id_list = religion_video_id_list[1:]

    # 按照排序给定分数
    new_result = {}
    step = 100 / (len(new_video_ids) * 2)
    for i, video_id in enumerate(new_video_ids):
        score = 100 - i * step
        new_result[int(video_id)] = score
    # 写入新的key中
    redis_helper.add_data_with_zset(key_name=new_key_name, data=new_result, expire_time=2 * 24 * 3600)
    return now_videos, religion_video_id_list


def merge_with_region(now_date, now_h, region, religion_video_id_list):
    initial_data_key = initial_param.get('data')
    initial_rule_key = initial_param.get('rule')
    new_data_key = new_param.get('data')
    new_rule_key = new_param.get('rule')
    now_videos = []
    # 地域小时级数据合并
    region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{initial_data_key}:{initial_rule_key}:" \
                        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    new_region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{new_data_key}:{new_rule_key}:" \
                            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    now_videos, religion_video_id_list = merge_process(initial_key_name=region_h_key_name,
                                                       new_key_name=new_region_h_key_name,
                                                       now_videos=now_videos,
                                                       religion_video_id_list=religion_video_id_list)
    # 地域24h数据合并
    region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{initial_data_key}:" \
                          f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    new_region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{new_data_key}:" \
                              f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    now_videos, religion_video_id_list = merge_process(initial_key_name=region_24h_key_name,
                                                       new_key_name=new_region_24h_key_name,
                                                       now_videos=now_videos,
                                                       religion_video_id_list=religion_video_id_list)

    # 24h筛选数据合并
    dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{initial_data_key}:" \
                        f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    new_dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{new_data_key}:" \
                            f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    now_videos, religion_video_id_list = merge_process(initial_key_name=dup2_24h_key_name,
                                                       new_key_name=new_dup2_24h_key_name,
                                                       now_videos=now_videos,
                                                       religion_video_id_list=religion_video_id_list)

    # 24h筛选后剩余数据合并
    dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{initial_data_key}:" \
                        f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    new_dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{new_data_key}:" \
                            f"{new_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
    now_videos, religion_video_id_list = merge_process(initial_key_name=dup3_24h_key_name,
                                                       new_key_name=new_dup3_24h_key_name,
                                                       now_videos=now_videos,
                                                       religion_video_id_list=religion_video_id_list)
    log_.info(f"region = {region} update end!")


def merge_videos(now_date, now_h):
    """将宗教视频插入到默认视频列表中"""
    # 获取宗教视频列表
    religion_videos = get_religion_videos(now_date=now_date)
    religion_video_id_list = [int(video_id) for video_id, _ in religion_videos]
    # 列表合并
    region_code_list = [code for region, code in config_.REGION_CODE.items()]
    task_list = [
        gevent.spawn(merge_with_region, now_date, now_h, region, religion_video_id_list)
        for region in region_code_list
    ]
    gevent.joinall(task_list)
    # 特殊城市视频数据准备
    for region, city_list in config_.REGION_CITY_MAPPING.items():
        t = [
            gevent.spawn(
                merge_with_region,
                now_date, now_h, city_code, religion_video_id_list
            )
            for city_code in city_list
        ]
        gevent.joinall(t)


if __name__ == '__main__':
    log_.info(f"laohaokan recommend data update start...")
    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')}")
    merge_videos(now_date, now_h)
    log_.info(f"laohaokan recommend data update end!")
    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"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}\n"
                 f"now_h: {now_h}\n"
                 f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}"
    )